Files
ViperC/App/lib/core/Gen/VarGenMixin.vp
T

1580 lines
94 KiB
Plaintext
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import lib.core.Gen.GenTypes as GenTypes
import lib.core.Gen.Generator as Generator
import lib.core.Gen.Type.TypeSpec as TSpec
from MatchGenMixin import SCOPE_MATCH
import lib.core.Gen.Translator as HT
from stdint import *
import ast
import mlirlite
import memhub
import string
import stdlib
import viperlib
# VarGenMixin - variable/scope symbol table Mixin
# ============================================================
# Constants (scope kinds; capacity constants are referenced via HT.).
# Scope kinds: module / function / block.
define SCOPE_MODULE = 0
define SCOPE_FUNCTION = 1
define SCOPE_BLOCK = 2
# Max variables per scope: the single source of truth is Translator.vp; this file always uses HT.MAX_VARS
# VarGenMixin - variable/scope symbol table Mixin
# ============================================================
# ============================================================
# M3:owned slife 绑定登记(函数级出口统一释放,见 ownership-and-new.md §4.2)
# ============================================================
# 每个"被 `new heap slife …` 绑定过"的变量槽登记两项:[变量槽 alloca, 所有权影子标志 alloca]。
# · 影子标志(i8,入口块置 0):1 = 该槽此刻持有的**确实**是编译器分配的对象;0 = 不是。
# 释放一律写成 `free(select(影子 != 0, load 槽, null))`(icmp + select,不建分支块):
# - 赋值写在 if/while 里而运行期没执行 → 影子仍是 0 → free(NULL),绝不 free 栈垃圾;
# - 槽里被写入过外来指针(`p = Foo()`)或借用别名(`q = p`)→ 该赋值处把影子清 0,
# 出口因此不会 free 外来指针;
# - 循环里反复 new → 赋值前影子仍是上一轮设的 1 → 先释放上一轮对象,不累积。
# · 为什么是**函数级**而不是块级:循环体/with/分支体不建立符号作用域,局部存储一律
# 入口块 alloca(函数级寿命),块级释放会让"块之后仍可达的变量"变成悬垂。
# · 登记区间是**栈式**的:[_g_slife_count, _g_slife_top) 属于当前函数。进入嵌套函数时
# 起点抬到当前栈顶(_SlifeReset),退出时**两个边界一起弹回**(_SlifeRestore 用 _g_slife_save
# 保存的旧值还原)——否则内层函数会在 0 号槽覆写外层登记项,且内层出口会把外层的对象一起 free 掉。
# · 表的存储必须**跨文件存活**:trans.Pool 是每文件池,Phase2 每翻完一个文件就 reset_to(mark2)
# (见 Phase2/__init__.vp)。若表落在这个池里,第二个文件起表体已是被回收的内存,后续
# 写入就是往活跃数据上砸(现场:viperc.exe 编译任何工程 access-violation on 0x0,
# 崩点每次都不同)。故与 _g_freed_slots 同口径,用 stdlib.malloc。
# 登记表容量(P5:128 → 256)。超限时 **fail-fast**(不再静默退化为"不登记 = 不释放"的泄漏)。
define _SLIFE_MAX = 256
# 区间保存栈深度(嵌套函数/方法体层数);只存两个边界值,每层 16 字节。
define _SLIFE_DEPTH_MAX = 64
# P2 willFree 失效槽表:_g_freed_slots[0.._g_freed_count) 是"已被标注调用消费过的变量槽 alloca"。
# **刻意不放进 VarEntry**(给热结构加字段会动布局、牵连整条变量注册路径)。
# 键 = alloca,槽在整次翻译中唯一,所以是"全局累积表",无需按函数 reset/restore。
define _FREED_MAX = 4096
_g_freed_slots: sizet* = None
_g_freed_count: int = 0
_g_slife_owned: sizet* = None
_g_slife_count: int = 0
_g_slife_top: int = 0
# 区间保存栈:[d*2+0]=进入第 d 层前的 _g_slife_count,[d*2+1]=进入前的 _g_slife_top。
_g_slife_save: sizet* = None
_g_slife_depth: int = 0
novirtual class VarGenMixin:
"""Variable/scope mixin"""
# init_vars - allocate and zero the VarEntry array
# ============================================================
Gen: Generator.Generator*
def init_vars(self, pool: memhub.MemBuddy*) -> GenTypes.VarEntry*:
"""Allocate and zero the variable table array"""
size: sizet = HT.MAX_VARS * GenTypes.VarEntry.__sizeof__()
vars_ptr: GenTypes.VarEntry* = pool.alloc(size)
if vars_ptr is not None: string.memset(vars_ptr, 0, size)
return vars_ptr
# find_var - look up a name in the variable table
# ============================================================
def find_var(self, vars_ptr: GenTypes.VarEntry*, var_count: int, name: str) -> mlirlite.Value*:
"""Look up by name in the variable table"""
if name is None or vars_ptr is None: return None
entry_size: sizet = GenTypes.VarEntry.__sizeof__()
for i in range(int(var_count)):
entry_addr: uint64 = uint64(vars_ptr) + i * entry_size
entry: GenTypes.VarEntry* = GenTypes.VarEntry*(entry_addr)
if entry.Name is not None and entry.Used != 0:
# Fast path: strcmp equality implies equal first chars, so skip strcmp when they differ
if entry.Name[0] == name[0] and string.strcmp(entry.Name, name) == 0: return entry.Alloca
return None
# find_var_entry - look up a name in the variable table, returning the VarEntry (with AnnotTy)
# ============================================================
def find_var_entry(self, vars_ptr: GenTypes.VarEntry*, var_count: int, name: str) -> GenTypes.VarEntry*:
"""Look up by name in the variable table, return the VarEntry or None"""
if name is None or vars_ptr is None: return None
entry_size: sizet = GenTypes.VarEntry.__sizeof__()
for i in range(int(var_count)):
entry_addr: uint64 = uint64(vars_ptr) + i * entry_size
entry: GenTypes.VarEntry* = GenTypes.VarEntry*(entry_addr)
if entry.Name is not None and entry.Used != 0:
if string.strcmp(entry.Name, name) == 0: return entry
return None
# lookup_var_entry - walk from the current scope upward, returning the VarEntry
# ============================================================
def lookup_var_entry(self, symtab: GenTypes.SymbolTable*, name: str) -> GenTypes.VarEntry*:
"""Walk up from the current scope to find a variable, return the VarEntry or None"""
if symtab is None or name is None: return None
scope: GenTypes.Scope* = symtab.Current
while scope is not None:
result: GenTypes.VarEntry* = self.find_var_entry(scope.Vars, scope.VarCount, name)
if result is not None: return result
scope = scope.Parent
return None
# set_var_annot_class_name - set a variable's original type-annotation class name.
# Stores it in Spec.SourceName so method dispatch can recover the struct type when alloca is Ptr(i8).
def set_var_annot_class_name(self, symtab: GenTypes.SymbolTable*, name: str, class_name: str) -> int:
"""Set a variable's original type-annotation class name into Spec.SourceName, return 0=success / 1=failure"""
if symtab is None or name is None: return 1
entry: GenTypes.VarEntry* = self.lookup_var_entry(symtab, name)
if entry is None: return 1
if class_name is None:
# No annotation name: do not write (VarEntry slots are not reused, so the original value is always None).
return 0
arena_sn: TSpec.TypeSpecArena* = self.Gen.TypeGenBase.ensure_spec_arena()
if arena_sn is None: return 1
if entry.Spec is None:
entry.Spec = TSpec.NewTypeSpec(arena_sn, TSpec.TS_UNKNOWN, 0, TSpec.TS_SIGN_UNKNOWN)
if entry.Spec is None: return 1
sign: int = self.Gen.TypeGenCType.spec_sign_of_annot(class_name)
if sign != TSpec.TS_SIGN_UNKNOWN: entry.Spec = TSpec.SpecWithSign(arena_sn, entry.Spec, sign)
entry.Spec = TSpec.SpecWithSource(entry.Spec, class_name)
return 0
# var_annot_class_name - unified read entry for a variable entry's annotation class name (S4).
# The annotation side writes Spec.SourceName via SpecWithSource; all read points go through here.
def var_annot_class_name(self, entry: GenTypes.VarEntry*) -> str:
"""The variable entry's annotation class name (Spec.SourceName); None if unset"""
if entry is None: return None
return TSpec.SpecSourceName(entry.Spec)
# set_var_ptr_element - mark a variable with "pointer element" semantics.
# bytes*/str*/T*/64-bit scalar|CPtr degrade to i8*, but indexing must use 8-byte slot strides (ElemBits=64).
def set_var_ptr_element(self, symtab: GenTypes.SymbolTable*, name: str) -> int:
"""Mark a variable as ptr-element (Spec.ElemBits=64), return 0=success / 1=failure"""
if symtab is None or name is None: return 1
entry: GenTypes.VarEntry* = self.lookup_var_entry(symtab, name)
if entry is None: return 1
# A ptr-element variable's alloca is Ptr(Ptr(Int8)), so its value type implies Spec PtrLevel>=1.
# Violation means an upstream type error; fail-fast, since a silent pass would mis-index by 1 byte.
if entry.Spec is None or entry.Spec.PtrLevel < 1:
self.Gen.TypeGenErr.fatal_error(None, f"variable '{name}' is marked as ptr-element, but its Spec cannot carry pointer semantics", "InternalCompilerError")
return 1
entry.Spec = TSpec.SpecWithElemBits(entry.Spec, 64)
return 0
# var_is_ptr_element - whether a variable has "pointer element" semantics (Spec.ElemBits).
# True when PtrLevel>=1 and Elem.Bits==8 and ElemBits==64; i64*/str/bytes/char* do not match.
def var_is_ptr_element(self, entry: GenTypes.VarEntry*) -> int:
"""Whether the variable has ptr-element semantics (1=yes / 0=no)"""
if entry is None or entry.Spec is None: return 0
sp: TSpec.TypeSpec* = entry.Spec
if sp.PtrLevel < 1 or sp.Elem is None: return 0
if sp.Elem.Bits != 8: return 0
if sp.ElemBits == 64: return 1
return 0
# set_var_volatile - mark a variable volatile (annotation union contains t.CVolatile).
# Every load/store of it carries LLVM volatile, preventing hoisting/removal of busy-wait loads.
def set_var_volatile(self, symtab: GenTypes.SymbolTable*, name: str) -> int:
"""Mark a variable as volatile (VarEntry.IsVolatile=1), return 0=success / 1=failure"""
if symtab is None or name is None: return 1
entry: GenTypes.VarEntry* = self.lookup_var_entry(symtab, name)
if entry is None: return 1
entry.IsVolatile = 1
return 0
# var_is_volatile - whether a variable is volatile (unified read/write entry)
def var_is_volatile(self, entry: GenTypes.VarEntry*) -> int:
"""Whether the variable entry is volatile (1=yes / 0=no; 0 when entry is None)"""
if entry is None: return 0
return entry.IsVolatile
# set_var_pointee_volatile - mark the pointed-to object volatile (rawptr[t.CVolatile | T]).
# Dereferences (p[i] / p.field / c.Deref(p)) then use LLVM volatile load/store on the pointee.
def set_var_pointee_volatile(self, symtab: GenTypes.SymbolTable*, name: str) -> int:
"""Mark the variable's pointee volatile (VarEntry.PointeeVolatile=1), return 0=success / 1=failure"""
if symtab is None or name is None: return 1
entry: GenTypes.VarEntry* = self.lookup_var_entry(symtab, name)
if entry is None: return 1
entry.PointeeVolatile = 1
return 0
# var_pointee_is_volatile - whether the pointed-to object is volatile (unified dereference entry)
def var_pointee_is_volatile(self, entry: GenTypes.VarEntry*) -> int:
"""Whether the variable entry's pointee is volatile (1=yes / 0=no; 0 when entry is None)"""
if entry is None: return 0
return entry.PointeeVolatile
# set_var_const - mark a variable as a const object (annotation has t.CConst and the type is not a pointer).
# Assignment / augmented assignment / element assignment to it are then rejected (same rule as TPC).
def set_var_const(self, symtab: GenTypes.SymbolTable*, name: str) -> int:
"""Mark a variable as a const object (VarEntry.IsConst=1), return 0=success / 1=failure"""
if symtab is None or name is None: return 1
entry: GenTypes.VarEntry* = self.lookup_var_entry(symtab, name)
if entry is None: return 1
entry.IsConst = 1
return 0
# var_is_const - whether a variable is a const object (unified intercept entry)
def var_is_const(self, entry: GenTypes.VarEntry*) -> int:
"""Whether the variable entry is a const object (1=yes / 0=no; 0 when entry is None)"""
if entry is None: return 0
return entry.IsConst
# set_var_byte_order - register a local variable's byte order (VarEntry.ByteOrder) from the
# `T | t.BigEndian` / `T | t.LittleEndian` annotation union: 1=BigEndian / 2=LittleEndian.
# The value is only 1/2 when a marker is present, so the local load/store paths bswap only then.
def set_var_byte_order(self, symtab: GenTypes.SymbolTable*, name: str, byte_order: int) -> int:
"""Set a local variable's byte order (VarEntry.ByteOrder), return 0=success / 1=failure"""
if symtab is None or name is None: return 1
entry: GenTypes.VarEntry* = self.lookup_var_entry(symtab, name)
if entry is None: return 1
entry.ByteOrder = byte_order
return 0
# var_byte_order - whether a variable is big/little-endian (unified read entry for load/store).
def var_byte_order(self, entry: GenTypes.VarEntry*) -> int:
"""The variable entry's byte order (0=native / 1=BigEndian / 2=LittleEndian; 0 when None)"""
if entry is None: return 0
return entry.ByteOrder
# ============================================================
# 生命期标签读写(LifeTag,见 GenTypes define LIFE_*;架构见 ownership-and-new.md §11)
# ============================================================
# set_var_life - write a variable's current-value life tag.
def set_var_life(self, symtab: GenTypes.SymbolTable*, name: str, life: int) -> int:
"""Set a variable's LifeTag (VarEntry.Life), return 0=success / 1=failure"""
if symtab is None or name is None: return 1
entry: GenTypes.VarEntry* = self.lookup_var_entry(symtab, name)
if entry is None: return 1
entry.Life = life
return 0
# get_var_life - read a variable's life tag; LIFE_UNKNOWN when unregistered (conservative).
def get_var_life(self, symtab: GenTypes.SymbolTable*, name: str) -> int:
"""A variable's LifeTag; LIFE_UNKNOWN when the variable is unknown"""
if symtab is None or name is None: return GenTypes.LIFE_UNKNOWN
entry: GenTypes.VarEntry* = self.lookup_var_entry(symtab, name)
if entry is None: return GenTypes.LIFE_UNKNOWN
return entry.Life
# life_meet - control-flow merge: the shorter life wins (smaller tag = shorter).
def life_meet(self, a: int, b: int) -> int:
"""Meet of two life tags (min); used where branches join"""
return a if a < b else b
# life_capture - 快照当前作用域链上所有变量的 LifeTag(供控制流汇合 Meet)。
# 返回池上数组:buf[0]=count,其后每项 (entry_ptr, life) 各占一个 sizet 槽;失败返回 None。
def life_capture(self, trans: HT.Translator*) -> sizet*:
"""Snapshot every in-scope variable's LifeTag (control-flow merge support)"""
if trans is None or trans.SymTab is None: return None
pool: memhub.MemBuddy* = trans.Pool
if pool is None: return None
cnt: sizet = 0
sc_c: GenTypes.Scope* = trans.SymTab.Current
while sc_c is not None:
cnt += sizet(sc_c.VarCount)
sc_c = sc_c.Parent
buf: sizet* = sizet*(pool.alloc(8 * (2 * cnt + 2)))
if buf is None: return None
buf[0] = cnt
idx: sizet = 1
esz: uint64 = uint64(GenTypes.VarEntry.__sizeof__())
sc: GenTypes.Scope* = trans.SymTab.Current
while sc is not None:
if sc.Vars is not None and sc.VarCount > 0:
j: int = 0
while j < sc.VarCount:
ent: GenTypes.VarEntry* = GenTypes.VarEntry*(uint64(sc.Vars) + uint64(j) * esz)
buf[idx * 2] = sizet(uint64(ent))
buf[idx * 2 + 1] = sizet(ent.Life)
idx += 1
j += 1
sc = sc.Parent
return buf
# life_restore - 把快照写回各变量(让 else 分支从 if 之前的状态开始)。
def life_restore(self, buf: sizet*) -> int:
"""Restore LifeTag values from a snapshot; 0=ok / 1=no snapshot"""
if buf is None: return 1
cnt: sizet = buf[0]
i: sizet = 1
while i <= cnt:
ent: GenTypes.VarEntry* = GenTypes.VarEntry*(uint64(buf[i * 2]))
if ent is not None: ent.Life = int(buf[i * 2 + 1])
i += 1
return 0
# life_join - 当前值与快照逐项 Meet(汇合:任一路径为短命即取短命)。
def life_join(self, buf: sizet*) -> int:
"""Meet the current LifeTags with a snapshot; 0=ok / 1=no snapshot"""
if buf is None: return 1
cnt: sizet = buf[0]
i: sizet = 1
while i <= cnt:
ent: GenTypes.VarEntry* = GenTypes.VarEntry*(uint64(buf[i * 2]))
if ent is not None:
ent.Life = self.life_meet(ent.Life, int(buf[i * 2 + 1]))
i += 1
return 0
# add_var - add a variable to the variable table
# ============================================================
def add_var(self, vars_ptr: GenTypes.VarEntry*, var_count: int, name: str, alloca: mlirlite.Value*) -> int:
"""Add a variable to the variable table"""
if name is None or alloca is None or vars_ptr is None: return 1
if var_count >= HT.MAX_VARS: return 1
entry_size: sizet = GenTypes.VarEntry.__sizeof__()
entry_addr: uint64 = uint64(vars_ptr) + var_count * entry_size
entry: GenTypes.VarEntry* = GenTypes.VarEntry*(entry_addr)
# Deep-copy the variable name into stdlib (survives pool reset): storing the AST Name.id pointer lets
# later allocations overwrite it, so find_var's strcmp silently misses and rebuilds a degenerate alloca.
name_len: sizet = string.strlen(name)
name_copy: bytes = stdlib.malloc(name_len + 1)
if name_copy is None: return 1
string.strcpy(name_copy, name)
entry.Name = name_copy
entry.Alloca = alloca
entry.Used = 1
# Explicitly zero the const flag (do not rely on the entry's zero initialization): set_var_const
# sets it to 1 only for real const objects, so zeroing avoids false positives.
entry.IsConst = 0
# 生命期标签显式初始化:未登记 = LIFE_UNKNOWN(保守放行)。
entry.Life = GenTypes.LIFE_UNKNOWN
# 字节序显式初始化:未标注 = 0 (native),只有 set_var_byte_order 才置 1/2(避免槽复用残留)。
entry.ByteOrder = 0
# [S3 dual write] Value-type Spec (alloca.Ty is Ptr(value type), so take its pointee)
entry.Spec = self._spec_of_var_alloca(alloca)
return 0
# ---- P2 willFree 失效槽表 API(不改 VarEntry 布局) ----
def _FreedMark(self, slot: mlirlite.Value*) -> int:
"""把变量槽标为"已失效"(被 @life(willFree(...)) 标注的调用消费过);1=已标记 / 0=未处理"""
global _g_freed_slots, _g_freed_count
if slot is None: return 0
if self._FreedCheck(slot) != 0: return 1
if _g_freed_slots is None:
_g_freed_slots = sizet*(stdlib.malloc(_FREED_MAX * 8))
if _g_freed_slots is None: return 0
if _g_freed_count >= _FREED_MAX:
self.Gen.TypeGenErr.fatal_error(None, "released-variable table full (capacity 4096): too many released variables in one translation", "LifetimeError")
return 0
_g_freed_slots[_g_freed_count] = sizet(slot)
_g_freed_count += 1
return 1
def _FreedCheck(self, slot: mlirlite.Value*) -> int:
"""该变量槽是否已被标为失效;1=已失效(其后再读/取址/传参即 LifetimeError)/ 0=有效"""
if slot is None or _g_freed_slots is None: return 0
i: int = 0
while i < _g_freed_count:
if uint64(_g_freed_slots[i]) == uint64(slot): return 1
i += 1
return 0
def _FreedClearSlot(self, slot: mlirlite.Value*) -> int:
"""清除某槽的失效态(对同名变量重新赋值后调用);1=清除过 / 0=本来不在表里"""
global _g_freed_count
if slot is None or _g_freed_slots is None: return 0
i: int = 0
while i < _g_freed_count:
if uint64(_g_freed_slots[i]) == uint64(slot):
_g_freed_count -= 1
_g_freed_slots[i] = _g_freed_slots[_g_freed_count]
return 1
i += 1
return 0
def clear_var_freed(self, symtab: GenTypes.SymbolTable*, name: str) -> int:
"""P2 willFree:对同名变量重新赋值后清除失效态;1=清除过 / 0=无该项"""
if symtab is None or name is None: return 0
return self._FreedClearSlot(self.lookup_var(symtab, name))
# ---- P2 willFree:失效槽表的控制流汇合(与 life_capture/life_restore/life_join 同点位) ----
# 背景:失效表是"平坦、单调"的全局表(_FreedMark 只插入),原本无分支合流语义,
# 于是"分支体内 free(x)"会把失效态泄漏到合流之后 → `if c: free(p); return` 之后的
# `return p` 误报。这里给 if/while/for 补上快照/复位/求交,口径 = "只有所有到达合流的
# 路径都失效,才算失效"(求交),与 life_* 的 Meet 同点位、同生命周期(pool 快照)。
def freed_capture(self, trans: HT.Translator*) -> sizet*:
"""快照当前失效槽表:buf[0]=count,其后每项一个失效槽指针;失败返回 None"""
global _g_freed_slots, _g_freed_count
if trans is None or trans.Pool is None: return None
pool: memhub.MemBuddy* = trans.Pool
n: sizet = sizet(_g_freed_count)
buf: sizet* = sizet*(pool.alloc((n + 1) * 8))
if buf is None: return None
buf[0] = n
i: sizet = 0
while i < n:
buf[1 + i] = _g_freed_slots[i]
i += 1
return buf
def freed_restore(self, buf: sizet*) -> int:
"""把失效槽表复位到快照(分支出口 / 单侧到达合流时用);0=完成 / 1=无快照"""
global _g_freed_slots, _g_freed_count
if buf is None: return 1
n: sizet = buf[0]
if n > 0:
if _g_freed_slots is None:
_g_freed_slots = sizet*(stdlib.malloc(_FREED_MAX * 8))
if _g_freed_slots is None: return 1
i: sizet = 0
while i < n:
_g_freed_slots[i] = buf[1 + i]
i += 1
_g_freed_count = int(n)
return 0
def freed_clear(self) -> None:
"""进入新函数体:清空失效槽表(函数作用域复位)。
表按 **alloca 槽地址** 记录(_FreedCheck 比较槽指针),而栈槽地址会被后续函数复用
(前帧已销毁)⇒ 若跨函数保留,上一函数的失效态会误加到新函数的同名/同址变量上
(现场:ExprGenCls._resolve_doc_symbol_attr 首行读形参 out_sym 即报 "already released",
而该函数体内根本没有 willFree 调用——是上一个函数的栈槽地址被复用)。
故每个函数体入口清零;嵌套函数体结束后不恢复外层失效态(仅为少报,绝非误报)。
"""
global _g_freed_count
_g_freed_count = 0
def freed_join(self, buf: sizet*) -> int:
"""合流:当前失效槽表与快照**求交**(只有两路都失效才算失效);0=完成 / 1=无快照"""
global _g_freed_slots, _g_freed_count
if buf is None: return 1
n: sizet = buf[0]
out: int = 0
i: sizet = 0
while i < sizet(_g_freed_count):
if _g_freed_slots is None: break
s: sizet = _g_freed_slots[i]
j: sizet = 0
keep: int = 0
while j < n:
if buf[1 + j] == s: keep = 1
if keep != 0: break
j += 1
if keep != 0:
_g_freed_slots[out] = s
out += 1
i += 1
_g_freed_count = out
return 0
# _spec_of_var_alloca - build a TypeSpec from the alloca's value type (S3 dual write).
# alloca.Ty is Ptr(var_ty), so the pointee is the value type; None means failure or an unsupported type.
def _spec_of_var_alloca(self, alloca: mlirlite.Value*) -> TSpec.TypeSpec*:
"""Build a TypeSpec from an alloca's value type; None if not applicable or failed"""
if alloca is None or alloca.Ty is None: return None
ty: mlirlite.LLVMType* = alloca.Ty
if self.Gen.ExprGenBase.is_ptr_type(ty) != 0: ty = ty.Pointee
arena: TSpec.TypeSpecArena* = self.Gen.TypeGenBase.ensure_spec_arena()
if arena is None: return None
return TSpec.SpecFromLLVMType(arena, ty)
# _alloca_at_entry - emit an alloca in the function entry block (ensures dominance).
# LLVM IR requires every instruction to dominate its uses, so all allocas go in the entry block.
def _alloca_at_entry(self, builder: mlirlite.IRBuilder*, ty: mlirlite.LLVMType*) -> mlirlite.Value*:
"""Emit an alloca in the function entry block before the terminator, return the alloca value."""
if builder is None or ty is None: return None
return mlirlite.build_alloca_at_entry(builder, ty)
# get_or_create_var - find or create a variable alloca (legacy-compatible)
def get_or_create_var(self, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, vars_ptr: GenTypes.VarEntry*, var_count: int, name: str, ty: mlirlite.LLVMType*) -> mlirlite.Value*:
"""Find or create a variable alloca"""
existing: mlirlite.Value* = self.find_var(vars_ptr, var_count, name)
if existing is not None: return existing
alloca: mlirlite.Value* = self._alloca_at_entry(builder, ty)
if alloca is None: return None
if self.add_var(vars_ptr, var_count, name, alloca) != 0: return None
return alloca
# _create_scope - create a new scope node (internal helper)
# ============================================================
def _create_scope(self, pool: memhub.MemBuddy*, parent: GenTypes.Scope*, kind: int) -> GenTypes.Scope*:
"""Create and initialize a scope node"""
scope: GenTypes.Scope* = pool.alloc(GenTypes.Scope.__sizeof__())
if scope is None: return None
string.memset(scope, 0, GenTypes.Scope.__sizeof__())
scope.Parent = parent
scope.Vars = self.init_vars(pool)
scope.VarCount = 0
scope.Kind = kind
if scope.Vars is None: return None
return scope
# init_symbol_table - create the symbol table (with the module-level root scope)
# ============================================================
def init_symbol_table(self, pool: memhub.MemBuddy*) -> GenTypes.SymbolTable*:
"""Create and initialize the symbol table with a module-level root scope"""
if pool is None: return None
symtab: GenTypes.SymbolTable* = pool.alloc(GenTypes.SymbolTable.__sizeof__())
if symtab is None: return None
string.memset(symtab, 0, GenTypes.SymbolTable.__sizeof__())
symtab.Pool = pool
# Create the root scope (module level)
root: GenTypes.Scope* = self._create_scope(pool, None, SCOPE_MODULE)
if root is None: return None
symtab.Root = root
symtab.Current = root
return symtab
# enter_scope - enter a new scope
# ============================================================
def enter_scope(self, symtab: GenTypes.SymbolTable*, kind: int) -> GenTypes.Scope*:
"""Enter a new scope and return it"""
if symtab is None: return None
pool: memhub.MemBuddy* = symtab.Pool
scope: GenTypes.Scope* = self._create_scope(pool, symtab.Current, kind)
if scope is None: return None
symtab.Current = scope
return scope
# exit_scope - exit the current scope
# ============================================================
def exit_scope(self, symtab: GenTypes.SymbolTable*):
"""Exit the current scope and restore its parent"""
if symtab is not None and symtab.Current is not None: symtab.Current = symtab.Current.Parent
# define_var - define a variable in the current scope
# ============================================================
def define_var(self, symtab: GenTypes.SymbolTable*, name: str, alloca: mlirlite.Value*) -> int:
"""Define a variable in the current scope, return 0 on success"""
if symtab is None or name is None or alloca is None: return 1
scope: GenTypes.Scope* = symtab.Current
if scope is None: return 1
ret: int = self.add_var(scope.Vars, scope.VarCount, name, alloca)
if ret == 0: scope.VarCount = scope.VarCount + 1
return ret
# define_module_var - define a variable in the module scope
# ============================================================
def define_module_var(self, symtab: GenTypes.SymbolTable*, name: str, alloca: mlirlite.Value*) -> int:
"""Define a variable in the module scope, return 0 on success"""
if symtab is None or name is None or alloca is None: return 1
scope: GenTypes.Scope* = symtab.Root
if scope is None: return 1
ret: int = self.add_var(scope.Vars, scope.VarCount, name, alloca)
if ret == 0: scope.VarCount = scope.VarCount + 1
return ret
# lookup_var - walk from the current scope upward looking for a variable
# ============================================================
def lookup_var(self, symtab: GenTypes.SymbolTable*, name: str) -> mlirlite.Value*:
"""Walk up from the current scope to find a variable, return its alloca or None"""
if symtab is None or name is None: return None
scope: GenTypes.Scope* = symtab.Current
while scope is not None:
result: mlirlite.Value* = self.find_var(scope.Vars, scope.VarCount, name)
if result is not None: return result
scope = scope.Parent
return None
# lookup_current - look up a variable only in the current scope
# ============================================================
def lookup_current(self, symtab: GenTypes.SymbolTable*, name: str) -> mlirlite.Value*:
"""Look up a variable in the current scope only"""
if symtab is None or name is None: return None
scope: GenTypes.Scope* = symtab.Current
if scope is None: return None
return self.find_var(scope.Vars, scope.VarCount, name)
# ============================================================
# M3 owned slife 登记 API(§4.2 函数级 scope guard / §4.3 不许手动 free)
# ============================================================
def _SlifeReset(self, pool: memhub.MemBuddy*) -> int:
"""函数入口:保存外层区间两个边界、把本层起点抬到当前栈顶;返回令牌(-1=不可用,须原样交还 _SlifeRestore)"""
global _g_slife_owned, _g_slife_save, _g_slife_count, _g_slife_top, _g_slife_depth
if _g_slife_owned is None:
# 表体跨文件存活:必须用 stdlib.malloc,**不能**用 pool(trans.Pool 是每文件池,
# Phase2 每文件结束 reset_to 后表体即悬垂,后续写入会砸活跃数据)。
_g_slife_owned = sizet*(stdlib.malloc(_SLIFE_MAX * 16))
if _g_slife_owned is None: return -1
if _g_slife_save is None:
_g_slife_save = sizet*(stdlib.malloc(_SLIFE_DEPTH_MAX * 16))
if _g_slife_save is None: return -1
if _g_slife_depth >= _SLIFE_DEPTH_MAX:
self.Gen.TypeGenErr.fatal_error(None, "slife ownership save stack full (function nesting > 64)", "LifetimeError")
return -1
_g_slife_save[_g_slife_depth * 2] = sizet(_g_slife_count)
_g_slife_save[_g_slife_depth * 2 + 1] = sizet(_g_slife_top)
tok: int = _g_slife_depth
_g_slife_depth = _g_slife_depth + 1
_g_slife_count = _g_slife_top
return tok
def _SlifeRestore(self, tok: int) -> int:
"""函数出口:**两个边界一起弹回**进入前的值(不释放;释放由 _SlifeEmitFrees 先做)。
关键:必须把 _g_slife_top 还原成进入前的值。只还原 count 会让 top 随函数数量单调增长,
上一文件/上一函数的陈旧条目就一直留在 [_g_slife_count, _g_slife_top) 被扫描——
而每文件池 reset_to 会复用同样的地址,于是"陈旧条目"能与新 alloca 地址相同,
_SlifeShadow 返回悬垂的影子槽,_SlifeNoteStore 直接写坏内存。"""
global _g_slife_count, _g_slife_top, _g_slife_depth
if tok < 0 or _g_slife_save is None: return 0
if tok >= _g_slife_depth: return 0
_g_slife_count = int(_g_slife_save[tok * 2])
_g_slife_top = int(_g_slife_save[tok * 2 + 1])
_g_slife_depth = tok
return 1
def _SlifeShadow(self, alloca: mlirlite.Value*) -> mlirlite.Value*:
"""查已登记槽的所有权影子标志槽(未登记 → None)"""
if alloca is None or _g_slife_owned is None: return None
i: int = 0
while i < _g_slife_top:
if uint64(_g_slife_owned[i * 2]) == uint64(alloca):
if _g_slife_owned[i * 2 + 1] == 0: return None
return mlirlite.Value*(_g_slife_owned[i * 2 + 1])
i += 1
return None
def _SlifeOwned(self, alloca: mlirlite.Value*) -> int:
"""该槽是否登记为 slife 绑定(= 由编译器负责释放,§4.3 判据5);1=是 / 0=否"""
if self._SlifeShadow(alloca) is not None: return 1
return 0
def _SlifeEnsureOwned(self, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, alloca: mlirlite.Value*) -> mlirlite.Value*:
"""登记该槽(首次:在入口块建 i8 影子并置 0),返回影子标志槽;不可用 → None"""
global _g_slife_top
if alloca is None or _g_slife_owned is None: return None
exist: mlirlite.Value* = self._SlifeShadow(alloca)
if exist is not None: return exist
if _g_slife_top >= _SLIFE_MAX:
# P5:超限 **fail-fast**(容量见文件头的 _SLIFE_MAX,此处文案里的数字与之同步)。
# 静默返回 None 会让该对象永不登记 → 永不释放(纯泄漏),与"编译器自动管理"的承诺矛盾;
# 下面的 return None 只是流程完整(fatal_error 已终止进程)。
self.Gen.TypeGenErr.fatal_error(None, "slife ownership table full (capacity 256): too many live compiler-owned slife objects in one function", "LifetimeError")
return None
if pool is None or builder is None: return None
sh: mlirlite.Value* = self._alloca_at_entry(builder, mlirlite.Int8(pool))
if sh is None: return None
zero8: mlirlite.Value* = mlirlite.const_int(pool, 8, 0)
if zero8 is None: return None
# 入口块置 0(显式传第 4 参,不依赖 include 默认值补全):赋值语句可能在运行期没执行,
# 那种路径上影子必须保持"不拥有",否则会 free 栈垃圾。
mlirlite.build_store_at_entry(builder, zero8, sh, 0)
_g_slife_owned[_g_slife_top * 2] = sizet(alloca)
_g_slife_owned[_g_slife_top * 2 + 1] = sizet(sh)
_g_slife_top += 1
return sh
def _SlifeNoteStore(self, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, sh: mlirlite.Value*, owned: int) -> int:
"""把影子标志写成 owned(1=该槽此刻持有编译器分配的对象 / 0=不持有)"""
if pool is None or builder is None or sh is None: return 0
v: mlirlite.Value* = mlirlite.const_int(pool, 8, owned)
if v is None: return 0
mlirlite.build_store(builder, v, sh)
return 1
def _SlifeEmitDtorBranch(self, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, mod: mlirlite.LLVMModule*, trans: HT.Translator*, alloca: mlirlite.Value*, sh: mlirlite.Value*) -> int:
"""§3.6 析构调度:拥有该对象时先调 `__del__`(对齐 TPC MemoryOps._emit_local_heap_frees:先析构、再 free)。
用**条件分支**而不是 select:未拥有的路径上必须完全不进析构体(析构体里会解引用 self)。
类(含 MRO / 跨模块 stub)没定义 `__del__` 时**不加任何分支**,直接返回——常见路径零开销。
析构名的选择:与 TPC 的**释放**路径一致用 `__del__`(`del obj` 那条语句才是 `__delete__`)。"""
if pool is None or builder is None or mod is None or trans is None: return 0
if alloca is None or sh is None: return 0
if alloca.Ty is None or alloca.Ty.Pointee == 0: return 0
elem: mlirlite.LLVMType* = alloca.Ty.Pointee # 变量类型(T*)
if elem is None or self.Gen.ExprGenBase.is_ptr_type(elem) == 0: return 0
struct_ty: mlirlite.LLVMType* = elem.Pointee # T
if struct_ty is None or struct_ty.kind() != mlirlite.LLVMType.Struct: return 0
cls_nm: str = self.Gen.StructGen.get_class_name_by_type(pool, struct_ty)
if cls_nm is None: return 0
if self.Gen.ExprGenOps._resolve_dunder_by_class(pool, mod, cls_nm, "__del__", trans, struct_ty) is None: return 0
func: mlirlite.Function* = builder.Func
if func is None: return 0
zero8: mlirlite.Value* = mlirlite.const_int(pool, 8, 0)
if zero8 is None: return 0
shv: mlirlite.Value* = mlirlite.build_load(builder, mlirlite.Int8(pool), sh)
if shv is None: return 0
cond: mlirlite.Value* = mlirlite.build_icmp(builder, mlirlite.ICMP_NE, shv, zero8)
if cond is None: return 0
cur: mlirlite.Value* = mlirlite.build_load(builder, alloca.Ty.Pointee, alloca)
if cur is None: return 0
# 唯一标签:用 trans._label_counter(与 translate_if 同源,避免同函数内重名标签)
cnt: int = trans._label_counter
trans._label_counter = cnt + 1
name_buf: str = pool.alloc(32)
if name_buf is None: return 0
viperlib.snprintf(name_buf, 32, "slife.dtor.%d", cnt)
dtor_bb: mlirlite.BasicBlock* = mlirlite.create_block(pool, func, name_buf)
viperlib.snprintf(name_buf, 32, "slife.dtor.end.%d", cnt)
merge_bb: mlirlite.BasicBlock* = mlirlite.create_block(pool, func, name_buf)
if dtor_bb is None or merge_bb is None: return 0
mlirlite.build_cond_br(builder, cond, dtor_bb, merge_bb)
mlirlite.position_at_end(builder, dtor_bb)
# `%cur` 定义在分支块之前,支配 dtor_bb 与 merge_bb 两边 ✓
self.Gen.ExprGenOps._try_named_dunder(pool, builder, mod, cur, "__del__", trans, None, 0)
mlirlite.build_br(builder, merge_bb)
mlirlite.position_at_end(builder, merge_bb)
return 1
def _SlifeEmitGuardedFree(self, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, mod: mlirlite.LLVMModule*, trans: HT.Translator*, alloca: mlirlite.Value*, sh: mlirlite.Value*) -> int:
"""释放该槽**确实拥有**的对象:先按 §3.6 调度析构 `__del__`(若类定义了),再 `free`。
释放本体用 icmp + select 而不是分支块:释放点可能插在任意位置(每个 return、覆盖赋值处),
select 无分支、无合流,安全得多;sh=0 时取 null,等价 free(NULL),槽里的外来指针/未初始化
值绝不参与释放。析构那一步另有条件分支(见 _SlifeEmitDtorBranch)。"""
if pool is None or builder is None or alloca is None or sh is None: return 0
if alloca.Ty is None or alloca.Ty.Pointee == 0: return 0
# §3.6:先析构,再 free
self._SlifeEmitDtorBranch(pool, builder, mod, trans, alloca, sh)
zero8: mlirlite.Value* = mlirlite.const_int(pool, 8, 0)
if zero8 is None: return 0
shv: mlirlite.Value* = mlirlite.build_load(builder, mlirlite.Int8(pool), sh)
if shv is None: return 0
cond: mlirlite.Value* = mlirlite.build_icmp(builder, mlirlite.ICMP_NE, shv, zero8)
if cond is None: return 0
cur: mlirlite.Value* = mlirlite.build_load(builder, alloca.Ty.Pointee, alloca)
if cur is None: return 0
# sh=0 时 select 取 null,槽里的值(可能是未初始化的垃圾或外来指针)**不会**被释放
nulp: mlirlite.Value* = mlirlite.ConstNull(pool, alloca.Ty.Pointee, "null")
if nulp is None: return 0
sel: mlirlite.Value* = mlirlite.build_select(builder, cond, cur, nulp)
if sel is None: return 0
i8p: mlirlite.LLVMType* = mlirlite.Ptr(pool, mlirlite.Int8(pool))
c: mlirlite.Value* = sel
if sel.Ty is not None and uint64(sel.Ty) != uint64(i8p):
c = mlirlite.build_bitcast(builder, sel, i8p)
if c is None: return 0
if mod is not None:
# `free` 由模块前言(Translator.vp,与 malloc 同处)统一声明,这里只查表 + fail-fast:
# 绝不吐出指向**未声明符号**的 llvm.call——mlir-opt 只会报
# "does not reference a symbol in the current scope",离根因很远(现场见 Translator.vp 的 free 注释)。
if self.Gen.CallGenBase.find_func_in_module(mod, "free") is None:
self.Gen.TypeGenErr.fatal_error(None, "cannot release a `slife` object: the module has no declaration of the C library function 'free'", "TypeError")
r: mlirlite.Value* = mlirlite.build_call(builder, "free", c, 1, mlirlite.Void(pool), 0)
if r is None: return 0
return 1
def _SlifeDropOwnedUnlessSame(self, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, mod: mlirlite.LLVMModule*, trans: HT.Translator*, target_slot: mlirlite.Value*, src_slot: mlirlite.Value*) -> int:
"""P4:覆盖释放 target 的当前对象;若它与 src 的当前值相同(自别名 `q = p; q = p`)则整体跳过。
没有这个运行时判等,重复自别名赋值会把两个槽共同持有的对象释放掉(q 与 p 同时悬垂);
判等放在 IR 里做(编译期无法知道两个槽是否指向同一对象)。"""
if target_slot is None or src_slot is None or trans is None: return 0
sh: mlirlite.Value* = self._SlifeShadow(target_slot)
if sh is None or target_slot.Ty is None or target_slot.Ty.Pointee == 0: return 0
el_ty: mlirlite.LLVMType* = target_slot.Ty.Pointee
qv: mlirlite.Value* = mlirlite.build_load(builder, el_ty, target_slot)
pv: mlirlite.Value* = mlirlite.build_load(builder, el_ty, src_slot)
if qv is None or pv is None: return 0
same: mlirlite.Value* = mlirlite.build_icmp(builder, mlirlite.ICMP_EQ, qv, pv)
if same is None: return 0
cur_func: mlirlite.Function* = trans._cur_func
if cur_func is None: return 0
cnt: int = trans._label_counter
trans._label_counter = cnt + 1
nm_free: str = pool.alloc(32)
if nm_free is None: return 0
viperlib.snprintf(nm_free, 32, "slife.xfer.free.%d", cnt)
nm_end: str = pool.alloc(32)
if nm_end is None: return 0
viperlib.snprintf(nm_end, 32, "slife.xfer.end.%d", cnt)
free_bb = mlirlite.create_block(pool, cur_func, nm_free)
end_bb = mlirlite.create_block(pool, cur_func, nm_end)
if free_bb is None or end_bb is None: return 0
# 相同 → 直接到 end(不释放);不同 → 走常规"保护式释放"(含 §3.6 析构调度)
mlirlite.build_cond_br(builder, same, end_bb, free_bb)
mlirlite.position_at_end(builder, free_bb)
self._SlifeEmitGuardedFree(pool, builder, mod, trans, target_slot, sh)
mlirlite.build_br(builder, end_bb)
mlirlite.position_at_end(builder, end_bb)
return 0
def _SlifeSlotFirstSeen(self, idx: int, slot: uint64) -> int:
"""槽 slot 在 (idx, _g_slife_top) 里是否已经出现过;1=本次首次出现(该释放)/ 0=重复(跳过)"""
j: int = int(idx + 1)
while j < _g_slife_top:
if uint64(_g_slife_owned[j * 2]) == slot: return 0
j += 1
return 1
def _SlifeEmitFrees(self, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, mod: mlirlite.LLVMModule*, trans: HT.Translator*) -> int:
"""函数出口(所有 return / 隐式收尾):逆序释放**本函数**登记且仍拥有的 slife 对象。
防御性去重:同一变量槽若在表里出现多次(登记路径异常 / 旧条目残留),只发一次 free——
对同一对象 free 两次是堆损坏(0xC0000374),绝不能出。"""
if pool is None or builder is None or _g_slife_owned is None: return 0
i: int = _g_slife_top - 1
while i >= _g_slife_count:
slot: uint64 = uint64(_g_slife_owned[i * 2])
sh: mlirlite.Value* = mlirlite.Value*(_g_slife_owned[i * 2 + 1])
if sh is not None and self._SlifeSlotFirstSeen(int(i), slot) != 0:
self._SlifeEmitGuardedFree(pool, builder, mod, trans, mlirlite.Value*(_g_slife_owned[i * 2]), sh)
i -= 1
return 0
# lookup_module_var - look up a variable in the module scope
# ============================================================
def lookup_module_var(self, symtab: GenTypes.SymbolTable*, name: str) -> mlirlite.Value*:
"""Look up a variable in the module scope"""
if symtab is None or name is None: return None
scope: GenTypes.Scope* = symtab.Root
if scope is None: return None
return self.find_var(scope.Vars, scope.VarCount, name)
# is_module_scope - whether the current scope is module level.
# Module-level AnnAssign modifies a module variable; inside a function (without global) it creates a local.
def is_module_scope(self, symtab: GenTypes.SymbolTable*) -> int:
"""Whether the current scope is module level (Current == Root), 1=module / 0=function scope"""
if symtab is None: return 0
if symtab.Root is None or symtab.Current is None: return 0
# Compare pointers: module level when Current equals Root
if uint64(symtab.Current) == uint64(symtab.Root): return 1
return 0
# _lookup_through_match_scope - look up a variable, walking through match-case scopes
def _lookup_through_match_scope(self, symtab: GenTypes.SymbolTable*, name: str) -> mlirlite.Value*:
"""Walk up from the current scope through SCOPE_MATCH scopes only to find a variable alloca"""
if symtab is None or name is None: return None
scope: GenTypes.Scope* = symtab.Current
while scope is not None:
result: mlirlite.Value* = self.find_var(scope.Vars, scope.VarCount, name)
if result is not None: return result
if scope.Kind == SCOPE_MATCH:
scope = scope.Parent
continue
return None
return None
# get_or_create_sym - find or create a variable (in the current scope)
# ============================================================
def get_or_create_sym(self, symtab: GenTypes.SymbolTable*, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, name: str, ty: mlirlite.LLVMType*) -> mlirlite.Value*:
"""Find or create a variable alloca in the current scope"""
existing: mlirlite.Value* = self.lookup_current(symtab, name)
if existing is not None: return existing
# The SCOPE_MATCH isolation scope must be transparent to assignment:
existing = self._lookup_through_match_scope(symtab, name)
if existing is not None: return existing
alloca: mlirlite.Value* = self._alloca_at_entry(builder, ty)
if alloca is None: return None
if self.define_var(symtab, name, alloca) != 0: self.Gen.TypeGenErr.fatal_error(None, f"failed to register variable '{name}' (scope variable count exceeds the limit MAX_VARS)", "InternalCompilerError")
return alloca
# REPL mode (M2) - session variable slot read/write; entered only when trans.ReplCtx != None (slots hold the value at ctx + x.Offset).
# ---- repl_find_entry - look up a session table entry by name ----
def repl_find_entry(self, ctx: GenTypes.ReplCtx*, name: str) -> GenTypes.ReplVarEntry*:
"""Look up an entry by name in the session table; None if ctx is invalid or unregistered"""
if ctx is None or name is None: return None
# Magic check: avoid treating a non-ReplCtx pointer as a context (better not to intercept than to over-intercept)
if ctx.Marker != GenTypes.REPL_CTX_MARKER: return None
if ctx.Table is None or ctx.Count <= 0: return None
entry_size: sizet = GenTypes.ReplVarEntry.__sizeof__()
for i in range(int(ctx.Count)):
entry_addr: uint64 = uint64(ctx.Table) + i * entry_size
entry: GenTypes.ReplVarEntry* = GenTypes.ReplVarEntry*(entry_addr)
if entry.Name is not None and string.strcmp(entry.Name, name) == 0: return entry
return None
# ---- repl_in_frag_scope - whether the nearest function scope is the fragment function itself ----
def repl_in_frag_scope(self, symtab: GenTypes.SymbolTable*) -> int:
"""1=the nearest function scope is the fragment body itself (its parent is the module root); 0=otherwise."""
if symtab is None or symtab.Current is None or symtab.Root is None: return 0
scope: GenTypes.Scope* = symtab.Current
while scope is not None:
if scope.Kind == SCOPE_FUNCTION:
if scope.Parent is not None and uint64(scope.Parent) == uint64(symtab.Root): return 1
return 0
scope = scope.Parent
return 0
# ---- repl_ctx_ptr - load the fragment function's ctx parameter value (i8*) ----
def repl_ctx_ptr(self, trans: HT.Translator*, ctx: GenTypes.ReplCtx*, node: ast.AST*) -> mlirlite.Value*:
"""Load the snippet function's ctx parameter (alloca built by FuncGenDef); fail fast on error"""
if trans is None or ctx is None or ctx.CtxParam is None: return None
symtab: GenTypes.SymbolTable* = trans.SymTab
param_alloca: mlirlite.Value* = self.lookup_var(symtab, ctx.CtxParam)
if param_alloca is None or param_alloca.Ty is None or param_alloca.Ty.Pointee == 0:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: snippet function is missing the ctx parameter '{ctx.CtxParam}' (wrapper template and M2 calling convention disagree)", "InternalCompilerError")
return None
return mlirlite.build_load(trans._cur_builder, param_alloca.Ty.Pointee, param_alloca)
# ---- repl_in_frag_entry_scope - whether the nearest function scope is the fragment entry function ----
# Differs from repl_in_frag_scope (M3): promoted session functions also parent to module root; write paths use this.
def repl_in_frag_entry_scope(self, trans: HT.Translator*) -> int:
"""1=currently translating a snippet entry function body; 0=otherwise (including promoted session bodies)"""
if trans is None or trans._cur_func is None: return 0
if self.repl_in_frag_scope(trans.SymTab) == 0: return 0
cur_nm: str = mlirlite.function_get_name(trans._cur_func)
if cur_nm is None: return 0
pfx: str = GenTypes.REPL_ENTRY_PREFIX
pl: sizet = string.strlen(pfx)
if string.strncmp(cur_nm, pfx, pl) != 0: return 0
return 1
# ---- repl_find_func_entry - look up a session function table entry by name (M3) ----
def repl_find_func_entry(self, ctx: GenTypes.ReplCtx*, name: str) -> GenTypes.ReplFuncEntry*:
"""Look up by name in the session function table; None if ctx is invalid or unregistered (read-only)"""
if ctx is None or name is None: return None
if ctx.Marker != GenTypes.REPL_CTX_MARKER: return None
if ctx.FuncTable is None or ctx.FuncCount <= 0: return None
entry_size: sizet = GenTypes.ReplFuncEntry.__sizeof__()
for i in range(int(ctx.FuncCount)):
entry_addr: uint64 = uint64(ctx.FuncTable) + i * entry_size
entry: GenTypes.ReplFuncEntry* = GenTypes.ReplFuncEntry*(entry_addr)
if entry.Name is not None and string.strcmp(entry.Name, name) == 0: return entry
return None
# ---- repl_func_ty - signature token -> LLVM type (M3) ----
# Tokens come from the host parsing the define line; only iN/float/double/i8* are supported (others -> error).
def repl_func_ty(self, pool: memhub.MemBuddy*, tag: int, bits: int) -> mlirlite.LLVMType*:
"""Map a signature token to an LLVM type; None for unsupported tokens"""
if pool is None: return None
if tag == GenTypes.REPL_TY_INT: return self.repl_scalar_ty(pool, bits, 0)
if tag == GenTypes.REPL_TY_FLOAT: return self.repl_scalar_ty(pool, bits, 1)
if tag == GenTypes.REPL_TY_STR: return mlirlite.Ptr(pool, mlirlite.Int8(pool))
return None
# ---- repl_func_sig_label - signature token -> human-readable type name (for error messages) ----
def repl_func_sig_label(self, tag: int, bits: int) -> str:
"""Map a signature token to a display name; "?" if unknown (VOID displays as void)"""
if tag == GenTypes.REPL_TY_VOID: return "void"
return self.repl_tag_label(tag, bits)
# ---- repl_try_session_func_call - indirect call to a session function (M3 core) ----
# On a hit, load the address from the ctx slot, prepend ctx, coerce args to the table signature, and call_indirect (hit=1).
def repl_try_session_func_call(self, trans: HT.Translator*, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, mod: mlirlite.LLVMModule*, func_name: str, cargs: list[ast.AST*]*, can: int, kw_count: int, node: ast.AST*, hit: rawptr) -> mlirlite.Value*:
"""Session function call: return the call result on hit (void returns None but hit is set)"""
if trans is None or trans.ReplCtx is None or pool is None or builder is None: return None
if func_name is None or hit is None: return None
ctx: GenTypes.ReplCtx* = trans.ReplCtx
entry: GenTypes.ReplFuncEntry* = self.repl_find_func_entry(ctx, func_name)
if entry is None: return None
# Same rule as M2's read path: the fragment entry function or a promoted session function may call
# other session functions; nested closures (no ctx param) are rejected, letting the existing path report.
if self.repl_in_frag_scope(trans.SymTab) == 0: return None
hit[0] = 1
if kw_count != 0:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: session function '{func_name}' does not support keyword arguments (M3 supports positional arguments only)", "TypeError")
return None
if can != entry.ParamCount:
self.Gen.TypeGenErr.fatal_error( node, f"REPL: session function '{func_name}' expects {entry.ParamCount} positional arguments, but the call site passed {can} (M3 does not support default arguments or default completion)", "TypeError")
return None
ctx_val: mlirlite.Value* = self.repl_ctx_ptr(trans, ctx, node)
if ctx_val is None: return None
i8_ty: mlirlite.LLVMType* = mlirlite.Int8(pool)
i8_ptr_ty: mlirlite.LLVMType* = mlirlite.Ptr(pool, i8_ty)
# (1) Load the function address from the slot: the slot is i8** (Ptr(i8*)); load yields i8*
slot: mlirlite.Value* = self.repl_slot_ptr(pool, builder, ctx_val, entry.SlotOff, i8_ptr_ty)
if slot is None:
self.Gen.TypeGenErr.fatal_error(node, "REPL: failed to generate the session function pointer slot address", "InternalCompilerError")
return None
fn_i8: mlirlite.Value* = mlirlite.build_load(builder, i8_ptr_ty, slot)
if fn_i8 is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: failed to read the address of session function '{func_name}'", "InternalCompilerError")
return None
# (2) Resolve param types before building the signature (token -> LLVM type; unsupported -> error)
ptypes: rawptr = pool.alloc(8 * GenTypes.REPL_FUNC_PARAM_MAX)
if ptypes is None: return None
string.memset(ptypes, 0, 8 * GenTypes.REPL_FUNC_PARAM_MAX)
ai_chk: int
for ai_chk in range(int(entry.ParamCount)):
pty_chk: mlirlite.LLVMType* = self.repl_func_ty(pool, int(entry.ParamTags[ai_chk]), int(entry.ParamBits[ai_chk]))
if pty_chk is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: type token of parameter {ai_chk + 1} of session function '{func_name}' is not supported (M3 supports iN/float/double/i8*)", "TypeError")
return None
pty_slot: str* = (str*)(uint64(ptypes) + ai_chk * 8)
pty_slot[0] = pty_chk
# (3) Args: the first is this fragment's ctx; the rest are translated and coerced to the signature types
head: mlirlite.Value* = ctx_val
head.Next = None
tail: mlirlite.Value* = head
arg_count: int = 1
ai: int
for ai in range(int(entry.ParamCount)):
arg: ast.AST* = cargs.get(ai)
if arg is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: argument {ai + 1} of session function '{func_name}' is missing", "SyntaxError")
return None
pty_slot2: str* = (str*)(uint64(ptypes) + ai * 8)
pty: mlirlite.LLVMType* = (mlirlite.LLVMType*)(pty_slot2[0])
arg_val: mlirlite.Value* = self.Gen.ExprGen.translate_value(builder, pool, mod, arg, None, 0, trans, pty)
if arg_val is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: failed to translate argument {ai + 1} of session function '{func_name}'", "TypeError")
return None
coerced: mlirlite.Value* = self.Gen.ExprGenBase.coerce_to_type(builder, arg_val, pty, arg, trans)
if coerced is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: argument {ai + 1} of session function '{func_name}' is incompatible with the signature type", "TypeError")
return None
coerced.Next = None
tail.Next = coerced
tail = coerced
arg_count += 1
# (4) Function type: ret + (i8*, P1..Pn) (void uses mlirlite.Void; build_call_indirect returns None,
# so the host caller treats it as a void call).
ret_ty: mlirlite.LLVMType* = mlirlite.Void(pool)
if entry.RetTag != GenTypes.REPL_TY_VOID:
ret_ty = self.repl_func_ty(pool, entry.RetTag, entry.RetBits)
if ret_ty is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: return type token of session function '{func_name}' is not supported (M3 supports void/iN/float/double/i8*)", "TypeError")
return None
# Param list nodes are mlirlite.ParamNode (new_param_node returns rawptr[ParamNode], and
# paramnode_set_next / Func params are rawptr[ParamNode] too) - previously mistyped as Function*.
fp_first: mlirlite.ParamNode* = None
fp_prev: mlirlite.ParamNode* = None
pi: int
for pi in range(int(arg_count)):
cur_ty: mlirlite.LLVMType* = i8_ptr_ty
if pi > 0:
pty_slot3: str* = (str*)(uint64(ptypes) + (pi - 1) * 8)
cur_ty = (mlirlite.LLVMType*)(pty_slot3[0])
pn: mlirlite.ParamNode* = mlirlite.new_param_node(pool, cur_ty)
if pn is not None:
if fp_first is None: fp_first = pn
if fp_prev is not None: mlirlite.paramnode_set_next(fp_prev, pn)
fp_prev = pn
func_ty: mlirlite.LLVMType* = mlirlite.Func(pool, ret_ty, fp_first, arg_count)
if func_ty is None: return None
func_ptr_ty: mlirlite.LLVMType* = mlirlite.Ptr(pool, func_ty)
fn_typed: mlirlite.Value* = mlirlite.build_bitcast(builder, fn_i8, func_ptr_ty)
if fn_typed is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: failed to convert the address of session function '{func_name}'", "InternalCompilerError")
return None
return mlirlite.build_call_indirect(builder, fn_typed, head, arg_count, ret_ty)
# ---- repl_slot_ptr - slot address (byte GEP from ctx + off, then bitcast to Ptr(val_ty)) ----
def repl_slot_ptr(self, pool: memhub.MemBuddy*, builder: mlirlite.IRBuilder*, ctx_val: mlirlite.Value*, off: int, val_ty: mlirlite.LLVMType*) -> mlirlite.Value*:
"""Return the slot pointer (Ptr(val_ty)) via a byte-offset i8 GEP then bitcast; None if args incomplete."""
if pool is None or builder is None or ctx_val is None or val_ty is None: return None
i8_ty: mlirlite.LLVMType* = mlirlite.Int8(pool)
target_ptr_ty: mlirlite.LLVMType* = mlirlite.Ptr(pool, val_ty)
off_val: mlirlite.Value* = mlirlite.const_int64(pool, int64(off))
if off_val is None: return None
base: mlirlite.Value* = mlirlite.build_gep(builder, i8_ty, ctx_val, off_val)
if base is None: return None
# Skip bitcast when types are equal: mlirlite types are not canonical, and LLVM rejects
# `bitcast i8* %x to i8*`. When val_ty is i8 the target is exactly i8*, so return the GEP result.
if base.Ty is not None and self.Gen.AssignGen._stg_ty_equal(pool, base.Ty, target_ptr_ty) != 0: return base
return mlirlite.build_bitcast(builder, base, target_ptr_ty)
# ---- repl_scalar_ty - bit width -> LLVM scalar type ----
def repl_scalar_ty(self, pool: memhub.MemBuddy*, bits: int, is_float: int) -> mlirlite.LLVMType*:
"""Restore an LLVM scalar type from a bit width; None for unsupported widths"""
if is_float != 0:
if bits == 32: return mlirlite.Float(pool)
if bits == 64: return mlirlite.Double(pool)
return None
if bits == 1: return mlirlite.Int1(pool)
if bits == 8: return mlirlite.Int8(pool)
if bits == 16: return mlirlite.Int16(pool)
if bits == 32: return mlirlite.Int32(pool)
if bits == 64: return mlirlite.Int64(pool)
return None
# ---- repl_llvm_ty - entry -> LLVM value type (structs re-resolved by name) ----
def repl_llvm_ty(self, pool: memhub.MemBuddy*, entry: GenTypes.ReplVarEntry*) -> mlirlite.LLVMType*:
"""Infer the entry's LLVM value type; None when untyped or a struct type cannot be re-resolved in this fragment."""
if entry is None or pool is None: return None
if entry.Kind == GenTypes.REPL_KIND_UNTYPED: return None
if entry.TyTag == GenTypes.REPL_TY_INT: return self.repl_scalar_ty(pool, entry.Bits, 0)
if entry.TyTag == GenTypes.REPL_TY_FLOAT: return self.repl_scalar_ty(pool, entry.Bits, 1)
if entry.TyTag == GenTypes.REPL_TY_STR: return mlirlite.Ptr(pool, mlirlite.Int8(pool))
if entry.TyTag == GenTypes.REPL_TY_PTR:
if entry.TyName is not None:
st: mlirlite.LLVMType* = self.Gen.StructGen.get_struct_type(entry.TyName)
if st is None: return None
return mlirlite.Ptr(pool, st)
ptee: mlirlite.LLVMType* = self.repl_scalar_ty(pool, entry.Bits, 0)
if ptee is None: ptee = mlirlite.Int8(pool)
return mlirlite.Ptr(pool, ptee)
if entry.TyTag == GenTypes.REPL_TY_STRUCT:
if entry.TyName is None: return None
return self.Gen.StructGen.get_struct_type(entry.TyName)
return None
# ---- repl_slot_ty - the value type stored in the slot ----
def repl_slot_ty(self, pool: memhub.MemBuddy*, entry: GenTypes.ReplVarEntry*) -> mlirlite.LLVMType*:
"""The value type stored in the slot equals the entry's value type (scalars/pointers/structs inline); None if it overflows the slot."""
if entry is None or pool is None: return None
vt: mlirlite.LLVMType* = self.repl_llvm_ty(pool, entry)
if vt is None: return None
if entry.TyTag == GenTypes.REPL_TY_STRUCT:
sz: int = self.repl_type_size(vt)
al: int = self.repl_type_align(vt)
if sz <= 0 or al <= 0 or sz > GenTypes.REPL_SLOT_SIZE or al > GenTypes.REPL_SLOT_SIZE: return None
return vt
# ---- repl_type_align / repl_type_size - struct value size/alignment (LLVM standard layout) ----
def repl_type_align(self, ty: mlirlite.LLVMType*, depth: int = 0) -> int:
"""Natural alignment bytes of a type; 0 if unknown (deep guards recursion depth)"""
if ty is None or depth > 64: return 0
ib: int = self.Gen.ExprGenBase.get_llvm_type_bits(ty)
if ib != 0:
if ib <= 8: return 1
if ib <= 16: return 2
if ib <= 32: return 4
return 8
fb: int = self.Gen.ExprGenBase.get_llvm_float_bits(ty)
if fb != 0:
if fb <= 16: return 2
if fb <= 32: return 4
return 8
if self.Gen.ExprGenBase.is_ptr_type(ty) != 0: return 8
elem_ty: mlirlite.LLVMType* = None
match ty:
case mlirlite.LLVMType.Array(elem_a, _cnt_a):
elem_ty = elem_a
case _:
pass
if elem_ty is not None: return self.repl_type_align(elem_ty, depth + 1)
fields: mlirlite.ParamNode* = mlirlite.get_struct_fields(ty)
if fields is None: return 0
cur: mlirlite.ParamNode* = fields
best: int = 1
while cur is not None:
fty: mlirlite.LLVMType* = mlirlite.paramnode_get_ty(cur)
fa: int = self.repl_type_align(fty, depth + 1)
if fa <= 0: return 0
if fa > best: best = fa
cur = mlirlite.paramnode_get_next(cur)
return best
def repl_type_size(self, ty: mlirlite.LLVMType*, depth: int = 0) -> int:
"""Type size in bytes under LLVM standard layout (fields aligned with tail padding); 0 if unknown."""
if ty is None or depth > 64: return 0
ib: int = self.Gen.ExprGenBase.get_llvm_type_bits(ty)
if ib != 0: return (ib + 7) // 8
fb: int = self.Gen.ExprGenBase.get_llvm_float_bits(ty)
if fb != 0: return fb // 8
if self.Gen.ExprGenBase.is_ptr_type(ty) != 0: return 8
elem_ty: mlirlite.LLVMType* = None
cnt: int = 0
match ty:
case mlirlite.LLVMType.Array(elem_s, cnt_s):
elem_ty = elem_s
cnt = cnt_s
case _:
pass
if elem_ty is not None:
es: int = self.repl_type_size(elem_ty, depth + 1)
if es <= 0 or cnt <= 0: return 0
return es * cnt
fields: mlirlite.ParamNode* = mlirlite.get_struct_fields(ty)
if fields is None: return 0
align_struct: int = self.repl_type_align(ty, depth + 1)
if align_struct <= 0: return 0
cur: mlirlite.ParamNode* = fields
off: int = 0
counted: int = 0
while cur is not None:
fty2: mlirlite.LLVMType* = mlirlite.paramnode_get_ty(cur)
fsz: int = self.repl_type_size(fty2, depth + 1)
fa2: int = self.repl_type_align(fty2, depth + 1)
if fsz <= 0 or fa2 <= 0: return 0
rem: int = off % fa2
if rem != 0: off += fa2 - rem
off += fsz
counted += 1
cur = mlirlite.paramnode_get_next(cur)
if counted <= 0: return 0
rem2: int = off % align_struct
if rem2 != 0: off += align_struct - rem2
return off
# ---- repl_seal - first-time typing (if already typed, verify the type tag matches) ----
def repl_seal(self, trans: HT.Translator*, entry: GenTypes.ReplVarEntry*, val_ty: mlirlite.LLVMType*, node: ast.AST*) -> int:
"""Type the entry by val_ty; return 0=success (idempotent hits included) / 1=unsupported or conflicting (fail-fast already raised)."""
if trans is None or entry is None or val_ty is None: return 1
perm: memhub.MemBuddy* = self.repl_perm_pool(trans)
if perm is None:
self.Gen.TypeGenErr.fatal_error(node, "REPL: session permanent pool is unavailable (not injected by the host)", "InternalCompilerError")
return 1
tag: int = 0
bits: int = 0
kind: int = GenTypes.REPL_KIND_SCALAR
tname: str = None
ib: int = self.Gen.ExprGenBase.get_llvm_type_bits(val_ty)
fb: int = self.Gen.ExprGenBase.get_llvm_float_bits(val_ty)
if ib != 0:
tag = GenTypes.REPL_TY_INT
bits = ib
elif fb != 0:
tag = GenTypes.REPL_TY_FLOAT
bits = fb
elif self.Gen.ExprGenBase.is_ptr_type(val_ty) != 0:
ptee: mlirlite.LLVMType* = val_ty.Pointee
pname: str = None
if ptee is not None: pname = mlirlite.get_struct_name(ptee)
if pname is not None:
tag = GenTypes.REPL_TY_PTR
bits = 64
tname = pname
else:
pb: int = 0
pf: int = 0
if ptee is not None:
pb = self.Gen.ExprGenBase.get_llvm_type_bits(ptee)
pf = self.Gen.ExprGenBase.get_llvm_float_bits(ptee)
if pb == 8:
tag = GenTypes.REPL_TY_STR
bits = 8
else:
tag = GenTypes.REPL_TY_PTR
if pb != 0: bits = pb
elif pf != 0: bits = pf
else: bits = 0
else:
sname: str = mlirlite.get_struct_name(val_ty)
if sname is not None:
# Struct value: inlined in the slot by declared type (no boxing). The slot capacity/alignment
# limit is REPL_SLOT_SIZE (16 bytes); exceeding it is a hard error, never silently truncated.
sz: int = self.repl_type_size(val_ty)
al: int = self.repl_type_align(val_ty)
if sz <= 0 or al <= 0:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: size/alignment of the struct value '{sname}' of session variable '{entry.Name}' is unknown" f" (cannot determine whether it fits in a {GenTypes.REPL_SLOT_SIZE} byte slot)", "TypeError")
return 1
if sz > GenTypes.REPL_SLOT_SIZE or al > GenTypes.REPL_SLOT_SIZE:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: struct value '{sname}' of session variable '{entry.Name}' is {sz} bytes and " f"aligned to {al} bytes, so it cannot fit in a {GenTypes.REPL_SLOT_SIZE} byte slot" f" (M2 struct values must be inlined into the slot; use an object pointer or heap allocation for large objects)", "TypeError")
return 1
tag = GenTypes.REPL_TY_STRUCT
bits = sz * 8
kind = GenTypes.REPL_KIND_SCALAR
tname = sname
else:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: the type of session variable '{entry.Name}' is not supported yet" f" (M2 supports int/bool/float/str/bytes/pointer/struct value)", "TypeError")
return 1
if entry.Kind == GenTypes.REPL_KIND_UNTYPED:
entry.Kind = kind
entry.TyTag = tag
entry.Bits = bits
entry.TyName = None
if tname is not None: entry.TyName = self.repl_perm_strdup(perm, tname)
return 0
if entry.Kind != kind or entry.TyTag != tag:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: session variable '{entry.Name}' is already typed as {self.repl_tag_label(entry.TyTag, entry.Bits)} and " f"cannot be assigned {self.repl_tag_label(tag, bits)} (restart the session to change its type)", "TypeError")
return 1
if tname is not None and entry.TyName is None: entry.TyName = self.repl_perm_strdup(perm, tname)
return 0
# ---- repl_perm_strdup - deep copy into PermPool (survives across fragments) ----
def repl_perm_strdup(self, pool: memhub.MemBuddy*, s: str) -> str:
"""Deep-copy s into pool (caller passes PermPool) for long-lived session storage; None on failure"""
if s is None or pool is None: return None
n: sizet = string.strlen(s)
buf: str = pool.alloc(n + 1)
if buf is None: return None
string.strcpy(buf, s)
return buf
# ---- repl_tag_label - type tag -> display name (same convention as the host echo) ----
def repl_tag_label(self, tag: int, bits: int) -> str:
"""Map TyTag/Bits to a human-readable type name (for error messages only)"""
if tag == GenTypes.REPL_TY_INT:
if bits == 1: return "bool"
if bits == 8: return "char"
if bits == 16: return "int16"
if bits == 32: return "int"
if bits == 64: return "int64"
return "int?"
if tag == GenTypes.REPL_TY_FLOAT:
if bits == 32: return "float"
if bits == 64: return "double"
return "float?"
if tag == GenTypes.REPL_TY_STR: return "str"
if tag == GenTypes.REPL_TY_PTR: return "ptr"
if tag == GenTypes.REPL_TY_STRUCT: return "struct"
return "?"
# ---- repl_ensure_entry - allocate a slot when an untyped session variable name is first seen ----
# Allocation happens during fragment translation on a local-table miss; a full table fails fast (no silent drop).
def repl_ensure_entry(self, trans: HT.Translator*, name: str, node: ast.AST*) -> GenTypes.ReplVarEntry*:
"""Get an entry by name, allocating and registering a slot if absent; None on failure (fail-fast already raised)."""
if trans is None or name is None: return None
ctx: GenTypes.ReplCtx* = trans.ReplCtx
if ctx is None or ctx.Marker != GenTypes.REPL_CTX_MARKER: return None
hit: GenTypes.ReplVarEntry* = self.repl_find_entry(ctx, name)
if hit is not None: return hit
if ctx.Table is None or ctx.Cap <= 0:
self.Gen.TypeGenErr.fatal_error(node, "REPL: session variable table is not initialized (host did not create ctx)", "InternalCompilerError")
return None
if ctx.Count >= ctx.Cap:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: session variables have reached the limit of {ctx.Cap} (ctx slots exhausted); " f"cannot allocate another slot for '{name}'; restart the session", "ResourceError")
return None
perm: memhub.MemBuddy* = self.repl_perm_pool(trans)
if perm is None:
self.Gen.TypeGenErr.fatal_error(node, "REPL: session permanent pool is unavailable (not injected by the host)", "InternalCompilerError")
return None
name_copy: str = self.repl_perm_strdup(perm, name)
if name_copy is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: failed to deep-copy the name of session variable '{name}'", "InternalCompilerError")
return None
entry_size: sizet = GenTypes.ReplVarEntry.__sizeof__()
entry: GenTypes.ReplVarEntry* = GenTypes.ReplVarEntry*(uint64(ctx.Table) + uint64(ctx.Count) * uint64(entry_size))
if entry is None: return None
entry.Name = name_copy
entry.TyName = None
entry.Offset = GenTypes.REPL_SLOT_BASE + ctx.Count * GenTypes.REPL_SLOT_SIZE
entry.Kind = GenTypes.REPL_KIND_UNTYPED
entry.TyTag = 0
entry.Bits = 0
ctx.Count = ctx.Count + 1
return entry
# ---- repl_perm_pool - the permanent pool for session data (survives across fragments) ----
def repl_perm_pool(self, trans: HT.Translator*) -> memhub.MemBuddy*:
"""The host-injected PermPool (ctx.Perm); None if there is no ctx or it was not injected"""
if trans is None or trans.ReplCtx is None: return None
return trans.ReplCtx.Perm
# ---- repl_emit_result - write this fragment's result into the ctx result area ----
def repl_emit_result(self, trans: HT.Translator*, ctx_val: mlirlite.Value*, val: mlirlite.Value*, tag: int, bits: int, node: ast.AST* = None) -> int:
"""Write val into the ctx result area per tag (see GenTypes result_kind); 0=success. Host reads only (kind, result_a, result_b)."""
if trans is None or ctx_val is None or val is None: return 1
pool: memhub.MemBuddy* = trans.Pool
builder: mlirlite.IRBuilder* = trans._cur_builder
if pool is None or builder is None: return 1
i64_ty: mlirlite.LLVMType* = mlirlite.Int64(pool)
zero64: mlirlite.Value* = mlirlite.const_int64(pool, 0)
if tag == GenTypes.REPL_TY_INT:
iv: mlirlite.Value* = self.Gen.ExprGenBase.coerce_to_type(builder, val, i64_ty, node, trans)
if iv is None: return 1
if self._repl_store_res_i32(trans, ctx_val, GenTypes.REPL_RES_KIND_OFF, GenTypes.REPL_RES_INT) != 0:
return 1
if self._repl_store_res_i64(trans, ctx_val, GenTypes.REPL_RES_A_OFF, iv) != 0: return 1
return self._repl_store_res_i64(trans, ctx_val, GenTypes.REPL_RES_B_OFF, zero64)
if tag == GenTypes.REPL_TY_FLOAT:
dv: mlirlite.Value* = self.Gen.ExprGenBase.coerce_to_type(builder, val, mlirlite.Double(pool), node, trans)
if dv is None: return 1
# double -> i64 bit pattern (bitcast, no numeric conversion): the host reinterprets the bits back to double
dbits: mlirlite.Value* = mlirlite.build_bitcast(builder, dv, i64_ty)
if dbits is None: return 1
if self._repl_store_res_i32(trans, ctx_val, GenTypes.REPL_RES_KIND_OFF, GenTypes.REPL_RES_FLOAT) != 0:
return 1
if self._repl_store_res_i64(trans, ctx_val, GenTypes.REPL_RES_A_OFF, dbits) != 0: return 1
return self._repl_store_res_i64(trans, ctx_val, GenTypes.REPL_RES_B_OFF, zero64)
if tag == GenTypes.REPL_TY_STR or tag == GenTypes.REPL_TY_PTR:
i8_ptr_ty: mlirlite.LLVMType* = mlirlite.Ptr(pool, mlirlite.Int8(pool))
pv: mlirlite.Value* = val
if val.Ty is not None and self.Gen.ExprGenBase.is_ptr_type(val.Ty) != 0 and self.Gen.AssignGen._stg_ty_equal(pool, val.Ty, i8_ptr_ty) == 0:
# Skip when already an i8 pointer (str/bytes/rawptr) - LLVM rejects same-type bitcast
pv = mlirlite.build_bitcast(builder, val, i8_ptr_ty)
if pv is None: return 1
# Pointer -> i64 (ptrtoint): the host reads it as an integer then inttoptrs it back to an address
pbits: mlirlite.Value* = self.Gen.ExprGenBase.coerce_to_type(builder, pv, i64_ty, node, trans)
if pbits is None: return 1
kind: int = GenTypes.REPL_RES_OBJ
if tag == GenTypes.REPL_TY_STR: kind = GenTypes.REPL_RES_STR
if self._repl_store_res_i32(trans, ctx_val, GenTypes.REPL_RES_KIND_OFF, kind) != 0:
return 1
if self._repl_store_res_i64(trans, ctx_val, GenTypes.REPL_RES_A_OFF, pbits) != 0: return 1
return self._repl_store_res_i64(trans, ctx_val, GenTypes.REPL_RES_B_OFF, zero64)
if tag == GenTypes.REPL_TY_STRUCT:
# Struct value: inlined in the session slot so the value itself has no address; the result area
# only reports the category (kind=4 and result_a=0), and the host prints <struct value>.
if self._repl_store_res_i32(trans, ctx_val, GenTypes.REPL_RES_KIND_OFF, GenTypes.REPL_RES_OBJ) != 0:
return 1
if self._repl_store_res_i64(trans, ctx_val, GenTypes.REPL_RES_A_OFF, zero64) != 0: return 1
return self._repl_store_res_i64(trans, ctx_val, GenTypes.REPL_RES_B_OFF, zero64)
return 1
# ---- _repl_store_res_i64 / _repl_store_res_i32 - write a fixed-width constant/value at a result offset ----
def _repl_store_res_i64(self, trans: HT.Translator*, ctx_val: mlirlite.Value*, off: int, val: mlirlite.Value*) -> int:
"""Write the i64 value to the ctx result area at offset off (0=success)"""
if trans is None or ctx_val is None or val is None: return 1
pool: memhub.MemBuddy* = trans.Pool
builder: mlirlite.IRBuilder* = trans._cur_builder
slot: mlirlite.Value* = self.repl_slot_ptr(pool, builder, ctx_val, off, mlirlite.Int64(pool))
if slot is None: return 1
mlirlite.build_store(builder, val, slot)
return 0
def _repl_store_res_i32(self, trans: HT.Translator*, ctx_val: mlirlite.Value*, off: int, val: int) -> int:
"""Write the i32 constant to the ctx result area at offset off (0=success)"""
if trans is None or ctx_val is None: return 1
pool: memhub.MemBuddy* = trans.Pool
builder: mlirlite.IRBuilder* = trans._cur_builder
if pool is None or builder is None: return 1
slot: mlirlite.Value* = self.repl_slot_ptr(pool, builder, ctx_val, off, mlirlite.Int32(pool))
if slot is None: return 1
cv: mlirlite.Value* = mlirlite.const_int32(pool, val)
if cv is None: return 1
mlirlite.build_store(builder, cv, slot)
return 0
# ---- repl_try_read - session variable read (returns the loaded value on a hit, else None) ----
def repl_try_read(self, trans: HT.Translator*, name: str, node: ast.AST*) -> mlirlite.Value*:
"""REPL session variable read: return the slot value on a hit (a box address for structs), else None."""
if trans is None or trans.ReplCtx is None or name is None: return None
entry: GenTypes.ReplVarEntry* = self.repl_find_entry(trans.ReplCtx, name)
if entry is None: return None
if self.repl_in_frag_scope(trans.SymTab) == 0: return None
if entry.Kind == GenTypes.REPL_KIND_UNTYPED:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: session variable '{name}' has not been assigned yet (untyped; first run a 'x = ...' or 'x: int = ...')", "NameError")
return None
pool: memhub.MemBuddy* = trans.Pool
slot_ty: mlirlite.LLVMType* = self.repl_slot_ty(pool, entry)
if slot_ty is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: the type " f"{self.repl_tag_label(entry.TyTag, entry.Bits)} of session variable '{name}' is not available in the current snippet" f" (cross-snippet type visibility belongs to M4)", "NameError")
return None
ctx_val: mlirlite.Value* = self.repl_ctx_ptr(trans, trans.ReplCtx, node)
if ctx_val is None: return None
slot: mlirlite.Value* = self.repl_slot_ptr(pool, trans._cur_builder, ctx_val, entry.Offset, slot_ty)
if slot is None:
self.Gen.TypeGenErr.fatal_error(node, "REPL: failed to generate the session slot address", "InternalCompilerError")
return None
return mlirlite.build_load(trans._cur_builder, slot_ty, slot)
# ---- repl_try_addr - session variable address-of (c.Addr(x) -> ctx slot address) ----
def repl_try_addr(self, trans: HT.Translator*, name: str, node: ast.AST*) -> mlirlite.Value*:
"""Session variable address-of: return the slot address (Ptr(value type)); None when unregistered, untyped, or outside the fragment body."""
if trans is None or trans.ReplCtx is None or name is None: return None
entry: GenTypes.ReplVarEntry* = self.repl_find_entry(trans.ReplCtx, name)
if entry is None: return None
if self.repl_in_frag_scope(trans.SymTab) == 0: return None
if entry.Kind == GenTypes.REPL_KIND_UNTYPED:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: session variable '{name}' has not been assigned yet (untyped), cannot take its address (c.Addr)", "NameError")
return None
pool: memhub.MemBuddy* = trans.Pool
slot_ty: mlirlite.LLVMType* = self.repl_slot_ty(pool, entry)
if slot_ty is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: the type of session variable '{name}' is not available in the current snippet (cross-snippet type visibility belongs to M4)", "NameError")
return None
ctx_val: mlirlite.Value* = self.repl_ctx_ptr(trans, trans.ReplCtx, node)
if ctx_val is None: return None
slot: mlirlite.Value* = self.repl_slot_ptr(pool, trans._cur_builder, ctx_val, entry.Offset, slot_ty)
if slot is None:
self.Gen.TypeGenErr.fatal_error(node, "REPL: failed to generate the session slot address", "InternalCompilerError")
return None
# The value is inlined in the slot (no boxing indirection), so the slot address is the value address
return slot
# ---- repl_store_slot - write a typed value into the slot (and into the result area) ----
def repl_store_slot(self, trans: HT.Translator*, entry: GenTypes.ReplVarEntry*, ctx_val: mlirlite.Value*, rhs_val: mlirlite.Value*, node: ast.AST*) -> int:
"""Coerce to the slot type, store into the slot, then write the result area; 0=success (fail-fast already raised on failure)."""
if trans is None or entry is None or ctx_val is None or rhs_val is None: return 1
pool: memhub.MemBuddy* = trans.Pool
builder: mlirlite.IRBuilder* = trans._cur_builder
slot_ty: mlirlite.LLVMType* = self.repl_slot_ty(pool, entry)
if slot_ty is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: the slot type of session variable '{entry.Name}' is not available" f" (the type is unresolvable in the current snippet, or the struct value exceeds the slot capacity)", "TypeError")
return 1
store_val: mlirlite.Value* = self.Gen.ExprGenBase.coerce_to_type(builder, rhs_val, slot_ty, node, trans)
if store_val is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: cannot convert the right-hand value of session variable '{entry.Name}' to" f" {self.repl_tag_label(entry.TyTag, entry.Bits)}", "CastError")
return 1
if self.Gen.ExprGenBase.is_ptr_type(slot_ty) != 0 and store_val.Ty is not None and self.Gen.ExprGenBase.is_ptr_type(store_val.Ty) != 0 and self.Gen.AssignGen._stg_ty_equal(pool, store_val.Ty, slot_ty) == 0:
# Pointer same-typing (the slot type and the RHS pointer may be different type instances but the same width).
# Skip when types are equal (including both i8*) - LLVM rejects same-type bitcast.
store_val = mlirlite.build_bitcast(builder, store_val, slot_ty)
if store_val is None:
self.Gen.TypeGenErr.fatal_error(node, "REPL: failed to unify the type of the session slot pointer", "CastError")
return 1
slot: mlirlite.Value* = self.repl_slot_ptr(pool, builder, ctx_val, entry.Offset, slot_ty)
if slot is None:
self.Gen.TypeGenErr.fatal_error(node, "REPL: failed to write the session slot", "InternalCompilerError")
return 1
mlirlite.build_store(builder, store_val, slot)
self.repl_emit_result(trans, ctx_val, store_val, entry.TyTag, entry.Bits, node)
return 0
# ---- repl_try_assign - session variable assignment (returns 1 on a hit) ----
def repl_try_assign(self, trans: HT.Translator*, name: str, rhs_val: mlirlite.Value*, rhs_node: ast.AST*, node: ast.AST*) -> int:
"""REPL session variable assignment: return 1 on a hit (registered and inside the fragment entry function), else 0."""
if trans is None or trans.ReplCtx is None or name is None or rhs_val is None: return 0
entry: GenTypes.ReplVarEntry* = self.repl_find_entry(trans.ReplCtx, name)
if entry is None: return 0
if self.repl_in_frag_entry_scope(trans) == 0: return 0
if rhs_val.Ty is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: the right-hand type of the assignment to session variable '{name}' is unknown", "TypeError")
return 1
if self.repl_seal(trans, entry, rhs_val.Ty, rhs_node) != 0: return 1
ctx_val: mlirlite.Value* = self.repl_ctx_ptr(trans, trans.ReplCtx, node)
if ctx_val is None: return 1
self.repl_store_slot(trans, entry, ctx_val, rhs_val, node)
return 1
# ---- repl_try_annassign - session variable annotated declaration/assignment (returns 1 on a hit) ----
def repl_try_annassign(self, trans: HT.Translator*, name: str, annot_node: ast.AST*, value_node: ast.AST*, node: ast.AST*) -> int:
"""REPL session variable annotated path (x: int = 1 / x: int, typing only): resolve the annotation, then type the entry and use it as the RHS target type."""
if trans is None or trans.ReplCtx is None or name is None: return 0
entry: GenTypes.ReplVarEntry* = self.repl_find_entry(trans.ReplCtx, name)
if entry is None: return 0
if self.repl_in_frag_entry_scope(trans) == 0: return 0
pool: memhub.MemBuddy* = trans.Pool
builder: mlirlite.IRBuilder* = trans._cur_builder
if annot_node is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: session variable '{name}' is missing an annotation", "SyntaxError")
return 1
# Same rule as ordinary AnnAssign: pre-register the generic specialization first so list[X] does not
# degrade to a bare base class (a bare list* makes later .get() misdispatch and truncate pointers).
self.Gen.AssignGen._prespecialize_generic_annotation(trans, pool, annot_node)
annot_ty: mlirlite.LLVMType* = self.Gen.TypeGenResolve.resolve_annotation_type(pool, annot_node, trans._imported_modules, trans._from_imports, trans, 0)
if annot_ty is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: failed to resolve the type annotation of session variable '{name}'" f" (the snippet has no import; only built-in names such as int/str/bytes/bool/float/double/char can be resolved)", "NameError")
return 1
if self.repl_seal(trans, entry, annot_ty, node) != 0: return 1
if value_node is None:
# Typing only (x: int): no value produced; the host zeroes this line's result area, so no result
return 1
slot_ty: mlirlite.LLVMType* = self.repl_slot_ty(pool, entry)
# Same rule as ordinary AnnAssign: set the context before translating the RHS (for bare generic construction inference)
trans.CurAssignAnnot = annot_node
trans.CurAssignNode = value_node
rhs_val: mlirlite.Value* = self.Gen.ExprGen.translate_value(builder, pool, trans.Module, value_node, None, 0, trans, slot_ty)
trans.CurAssignAnnot = None
trans.CurAssignNode = None
if rhs_val is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: failed to translate the initial value of session variable '{name}'", "TypeError")
return 1
ctx_val: mlirlite.Value* = self.repl_ctx_ptr(trans, trans.ReplCtx, node)
if ctx_val is None: return 1
self.repl_store_slot(trans, entry, ctx_val, rhs_val, node)
return 1
# ---- repl_try_augassign - session variable augmented assignment (returns 1 on a hit) ----
def repl_try_augassign(self, trans: HT.Translator*, name: str, op: int, rhs_node: ast.AST*, node: ast.AST*) -> int:
"""REPL session variable x += v: read slot, apply the binary op, write the slot and result area (return 1 on a hit)."""
if trans is None or trans.ReplCtx is None or name is None: return 0
entry: GenTypes.ReplVarEntry* = self.repl_find_entry(trans.ReplCtx, name)
if entry is None: return 0
if self.repl_in_frag_entry_scope(trans) == 0: return 0
pool: memhub.MemBuddy* = trans.Pool
builder: mlirlite.IRBuilder* = trans._cur_builder
if entry.TyTag == GenTypes.REPL_TY_STRUCT:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: session variable '{name}' is a struct value, augmented assignment is not supported", "TypeError")
return 1
cur_val: mlirlite.Value* = self.repl_try_read(trans, name, node)
if cur_val is None: return 1
# 与 `x = x op y` 同口径:RHS 在左值槽的 LLVM 类型(target_ty)上下文中翻译,使嵌套 binop /
# CDefine 的目标感知与常量适配生效,避免 REPL 会话变量的增强赋值与二元形态口径不一致
# (对齐 AssignGenMixin 的字段/下标/解引用/变量四个站点)。cur_val 已由 repl_try_read 成功
# 加载,故槽类型必可解析;缺失即内部错误 fail-fast(绝不退回 None 或 i32 兜底)。
slot_ty: mlirlite.LLVMType* = self.repl_slot_ty(pool, entry)
if slot_ty is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: the slot type of session variable '{name}' is not available", "TypeError")
return 1
rhs_val: mlirlite.Value* = self.Gen.ExprGen.translate_value(builder, pool, trans.Module, rhs_node, None, 0, trans, slot_ty)
if rhs_val is None:
self.Gen.TypeGenErr.fatal_error(node, "REPL: failed to translate the right-hand side of the augmented assignment", "TypeError")
return 1
result: mlirlite.Value* = self.Gen.AssignGen._apply_aug_op(pool, builder, op, cur_val, rhs_val, node, trans, node)
if result is None:
self.Gen.TypeGenErr.fatal_error(node, f"REPL: augmented assignment on session variable '{name}' failed op={op}", "TypeError")
return 1
# Write back to the slot (reuses the assignment path: typing already exists, so only convert and store)
return self.repl_try_assign(trans, name, result, node, node)
# ---- repl_expr_result - result pass-through for a bare expression statement in a fragment ----
def repl_expr_result(self, trans: HT.Translator*, value_node: ast.AST*) -> int:
"""In REPL mode, evaluate a bare expression statement (e.g. x + 1) and write the result area; 0=success."""
if trans is None or trans.ReplCtx is None or value_node is None: return 1
pool: memhub.MemBuddy* = trans.Pool
builder: mlirlite.IRBuilder* = trans._cur_builder
val: mlirlite.Value* = self.Gen.ExprGen.translate_value(builder, pool, trans.Module, value_node, None, 0, trans)
if val is None or val.Ty is None:
self.Gen.TypeGenErr.fatal_error(value_node, "REPL: failed to evaluate the expression snippet", "TypeError")
return 1
ctx_val: mlirlite.Value* = self.repl_ctx_ptr(trans, trans.ReplCtx, value_node)
if ctx_val is None: return 1
# Type classification uses the host's scratch entry (occupies no session slot, not in the table); reset to untyped each time
sh: GenTypes.ReplVarEntry* = trans.ReplCtx.Scratch
if sh is None:
self.Gen.TypeGenErr.fatal_error(value_node, "REPL: session is missing the shadow entry (host not initialized)", "InternalCompilerError")
return 1
sh.Kind = GenTypes.REPL_KIND_UNTYPED
sh.TyTag = 0
sh.Bits = 0
sh.TyName = None
if self.repl_seal(trans, sh, val.Ty, value_node) != 0: return 1
return self.repl_emit_result(trans, ctx_val, val, sh.TyTag, sh.Bits, value_node)