Files
ViperC/App/lib/StubGen/VpiReader.vp
T
TermiNexus a9ebb20790 "fix(compiler): vpi 返回注解缓存按需加载 + 动态扩容
1. _lookup_method_ret_ty_via_vpi 查缓存前按需读取
   {temp_dir}/{sha1[0:2]}/{sha1}.vpi 填充缓存,覆盖模块自身翻译时
   (同模块方法调用、定义在后)stub 尚未注入的场景
2. 缓存容量由固定 4096 改为 stdlib.realloc 翻倍动态扩容(上限 1M),
   根治写满后条目静默丢弃导致 JsonValue._pool_alloc 等查不到"
2026-08-08 02:24:12 +08:00

986 lines
45 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 t, c
from stdint import *
import ast
import string
import stdlib
import sys
import memhub
import llvmlite
import viperlib
import w32.fileio as fileio
import lib.core.VLogger as VLogger
import lib.core.Handles.HandlesType as HandlesType
import lib.core.Handles.HandlesStruct as HandlesStruct
import lib.core.Handles.HandlesExprCall as HandlesExprCall
import lib.core.Handles.HandlesFunctions as HandlesFunctions
import lib.core.Handles.HandlesTranslator as HT
# ============================================================
# 方法返回注解缓存(stdlib 持久)
#
# 背景:vpi tree 在"解析后立即"有效,但遍历处理到后续类的方法时,
# 深层 AST 节点(如 parse_args 的 returns 注解)可能已被覆盖损坏
# (kind() 读取异常),直接 resolve 得到垃圾类型(如 i8*)。
# 因此采用文本提取方案:ParseVpiIntoModule 读 .vpi stub 时,从
# vpi_buf 原始文本(在函数末尾才释放,始终有效)直接按行格式
# 提取 "def m(...) -> Ret:" 的返回注解字符串缓存到 stdlib(持久,
# 不受 pool 状态影响)。HandlesExprCall._lookup_method_ret_ty_via_vpi
# 按 sha1.类名.方法名 查表并用 _resolve_vpi_ret_annot_str 纯字符串
# 解析为 LLVMType,返回类型始终来自方法定义处,无白名单/无硬编码。
# ============================================================
VPI_RET_MAX: t.CDefine = 4096
VPI_KEY_LEN: t.CDefine = 224
VPI_RET_LEN: t.CDefine = 224
_g_vpi_ret_keys: bytes = None
_g_vpi_ret_strs: bytes = None
_g_vpi_ret_count: int = 0
_g_vpi_ret_cap: int = 0
# ============================================================
# _cache_vpi_return_annots_text - 从 vpi_buf 文本直接提取方法返回注解
#
# 完全不依赖 AST tree(tree 深层节点在后续处理中可能损坏),
# 直接按 vpi 生成文件的行格式提取:
# " def method(...) -> Ret: pass"(类内方法,缩进)
# "def method(...) -> Ret: pass"(模块级函数)
# "class X:"(跟踪当前类名)
# ============================================================
def _cache_vpi_return_annots_text(vpi_buf: bytes, src_sha1: str) -> int:
"""从 vpi_buf 文本提取方法返回注解字符串(stdlib 持久)"""
global _g_vpi_ret_keys, _g_vpi_ret_strs, _g_vpi_ret_count, _g_vpi_ret_cap
if vpi_buf is None or src_sha1 is None:
return 0
if _g_vpi_ret_keys is None:
_g_vpi_ret_keys = stdlib.malloc(VPI_RET_MAX * VPI_KEY_LEN)
_g_vpi_ret_strs = stdlib.malloc(VPI_RET_MAX * VPI_RET_LEN)
if _g_vpi_ret_keys is None or _g_vpi_ret_strs is None:
return 0
string.memset(_g_vpi_ret_keys, 0, VPI_RET_MAX * VPI_KEY_LEN)
string.memset(_g_vpi_ret_strs, 0, VPI_RET_MAX * VPI_RET_LEN)
_g_vpi_ret_cap = VPI_RET_MAX
sha1_len: t.CSizeT = string.strlen(src_sha1)
buf_len: t.CSizeT = string.strlen(vpi_buf)
cur_cls: bytes = None
pos: t.CSizeT = 0
line_no: int = 0
while pos < buf_len:
ls: t.CSizeT = pos
le: t.CSizeT = pos
while le < buf_len and vpi_buf[le] != '\n':
le += 1
line_len: t.CSizeT = le - ls
# 跳过前导空格,检查 class/def
i: t.CSizeT = ls
while i < le and vpi_buf[i] == ' ':
i += 1
if i < le:
# class X:
if line_len >= 6 and string.strncmp(vpi_buf + i, "class ", 6) == 0:
cstart: t.CSizeT = i + 6
cend: t.CSizeT = cstart
while cend < le and vpi_buf[cend] != ':' and vpi_buf[cend] != ' ':
cend += 1
if cend > cstart:
if cur_cls is not None:
stdlib.free(cur_cls)
cur_cls = stdlib.malloc(cend - cstart + 1)
if cur_cls is not None:
string.strncpy(cur_cls, vpi_buf + cstart, cend - cstart)
cur_cls[cend - cstart] = '\0'
# def name(...) -> Ret:
elif line_len >= 4 and string.strncmp(vpi_buf + i, "def ", 4) == 0:
dstart: t.CSizeT = i + 4
dend: t.CSizeT = dstart
while dend < le and vpi_buf[dend] != '(':
dend += 1
# 方法名 = [dstart, dend)
# 找 "->"
rpos: t.CSizeT = dstart
arrow_pos: t.CSizeT = le
while rpos < le - 1:
if vpi_buf[rpos] == '-' and vpi_buf[rpos + 1] == '>':
arrow_pos = rpos
break
rpos += 1
if arrow_pos < le and dend > dstart:
# 返回类型 = arrow_pos+2 跳过空格 到 ':' 前
rstart: t.CSizeT = arrow_pos + 2
while rstart < le and vpi_buf[rstart] == ' ':
rstart += 1
rend: t.CSizeT = rstart
while rend < le and vpi_buf[rend] != ':':
rend += 1
if rend > rstart:
# 构造 key 和 ret_str,存入缓存
mname_len: t.CSizeT = dend - dstart
rtype_len: t.CSizeT = rend - rstart
cls_len2: t.CSizeT = 0
if cur_cls is not None:
cls_len2 = string.strlen(cur_cls)
key_len: t.CSizeT = sha1_len + 1 + cls_len2 + 1 + mname_len
if key_len < VPI_KEY_LEN and rtype_len < VPI_RET_LEN:
# 方法名复制到临时缓冲(vpi_buf+dstart 非 null 结尾,不能直接 snprintf)
mname_buf: bytes = stdlib.malloc(mname_len + 1)
if mname_buf is not None:
string.strncpy(mname_buf, vpi_buf + dstart, mname_len)
mname_buf[mname_len] = '\0'
key_buf: bytes = stdlib.malloc(key_len + 1)
if key_buf is not None:
if cur_cls is not None:
viperlib.snprintf(key_buf, key_len + 1, "%s.%s.%s", src_sha1, cur_cls, mname_buf)
else:
viperlib.snprintf(key_buf, key_len + 1, "%s.%s", src_sha1, mname_buf)
# 去重:同一 key 已存在则跳过(同文件可能被多次加载)
dup_key: int = 0
dki: int
for dki in range(_g_vpi_ret_count):
if string.strcmp(_g_vpi_ret_keys + dki * VPI_KEY_LEN, key_buf) == 0:
dup_key = 1
break
if dup_key == 0:
grow_ok: int = 1
if _g_vpi_ret_count >= _g_vpi_ret_cap:
grow_ok = _grow_vpi_ret_cache()
if grow_ok != 0:
string.strcpy(_g_vpi_ret_keys + _g_vpi_ret_count * VPI_KEY_LEN, key_buf)
string.strncpy(_g_vpi_ret_strs + _g_vpi_ret_count * VPI_RET_LEN,
vpi_buf + rstart, rtype_len)
(_g_vpi_ret_strs + _g_vpi_ret_count * VPI_RET_LEN)[rtype_len] = '\0'
_g_vpi_ret_count += 1
stdlib.free(key_buf)
stdlib.free(mname_buf)
pos = le + 1
line_no += 1
if cur_cls is not None:
stdlib.free(cur_cls)
return 1
# ============================================================
# _grow_vpi_ret_cache - 扩容方法返回注解缓存(容量翻倍)
#
# 背景:VPI_RET_MAX 初始 4096,但 includes 模块量大(~90 个 stub、
# 每个几十个方法),缓存写满后条目被静默丢弃,导致已缓存模块的
# 方法(如 JsonValue._pool_alloc)查不到而 fail-fast 误报。
# 用 stdlib.realloc 翻倍扩容(上限 1M 条目,防御异常增长)。
# ============================================================
def _grow_vpi_ret_cache() -> int:
"""扩容返回注解缓存(容量翻倍),返回 1=成功 / 0=失败"""
global _g_vpi_ret_keys, _g_vpi_ret_strs, _g_vpi_ret_cap
if _g_vpi_ret_keys is None or _g_vpi_ret_cap <= 0:
return 0
new_cap: int = _g_vpi_ret_cap * 2
if new_cap > 1048576:
return 0
new_keys: bytes = stdlib.realloc(_g_vpi_ret_keys, new_cap * VPI_KEY_LEN)
if new_keys is None:
return 0
new_strs: bytes = stdlib.realloc(_g_vpi_ret_strs, new_cap * VPI_RET_LEN)
if new_strs is None:
return 0
_g_vpi_ret_keys = new_keys
_g_vpi_ret_strs = new_strs
_g_vpi_ret_cap = new_cap
return 1
# ============================================================
# _ensure_vpi_ret_annot_loaded - 按需加载模块的 vpi stub 返回注解缓存
#
# 背景:_cache_vpi_return_annots_text 仅在 stub 注入(LoadDepVpiIntoModule,
# 即"当前模块 import 了该模块")时填充缓存。但模块自身翻译时,自己的
# stub 尚未被任何依赖方读取,且自己的 text.ll 要翻译完成才落盘,导致
# "同模块方法调用(定义在后)"(如 json.__setitem__ 调用后定义的
# _pool_alloc)在 found_func/text.ll/vpi 缓存三类来源上全部未命中。
# 此函数在缓存未命中时直接读 {temp_dir}/{sha1[0:2]}/{sha1}.vpi
# (Converter 在 deps 阶段已生成全部 stub 文件),填充缓存后即可命中。
# 幂等:模块已缓存过则跳过,不重复读文件。
# ============================================================
def _ensure_vpi_ret_annot_loaded(src_sha1: str, temp_dir: str) -> int:
"""按需加载模块 vpi stub 的返回注解缓存,返回 1=已就绪 / 0=失败"""
if src_sha1 is None or temp_dir is None:
return 0
sha1_len: t.CSizeT = string.strlen(src_sha1)
if sha1_len < 2:
return 0
# 该模块已有缓存条目 → 跳过(幂等)
if _g_vpi_ret_keys is not None and _g_vpi_ret_count > 0:
pre_i: int
for pre_i in range(_g_vpi_ret_count):
pre_key: bytes = _g_vpi_ret_keys + pre_i * VPI_KEY_LEN
if string.strncmp(pre_key, src_sha1, sha1_len) == 0 and pre_key[sha1_len] == '.':
return 1
# 构造路径 {temp_dir}/{sha1[0:2]}/{sha1}.vpi(逐字符拼接,避免 %c 格式化问题)
td_len: t.CSizeT = string.strlen(temp_dir)
path_len: t.CSizeT = td_len + sha1_len + 12
vpi_path: bytes = stdlib.malloc(path_len)
if vpi_path is None:
return 0
string.strcpy(vpi_path, temp_dir)
vpi_path[td_len] = '/'
vpi_path[td_len + 1] = src_sha1[0]
vpi_path[td_len + 2] = src_sha1[1]
vpi_path[td_len + 3] = '/'
string.strcpy(vpi_path + td_len + 4, src_sha1)
string.strcpy(vpi_path + td_len + 4 + sha1_len, ".vpi")
f: fileio.File | t.CPtr = fileio.File(vpi_path, fileio.MODE.R)
if f.closed:
stdlib.free(vpi_path)
return 0
vpi_buf: bytes = stdlib.malloc(VPI_READ_BUF_SIZE)
if vpi_buf is None:
f.close()
stdlib.free(vpi_path)
return 0
bytes_read: LONG = f.read_all(vpi_buf, VPI_READ_BUF_SIZE)
f.close()
if bytes_read > 0:
if bytes_read < VPI_READ_BUF_SIZE:
vpi_buf[bytes_read] = 0
else:
vpi_buf[VPI_READ_BUF_SIZE - 1] = 0
_cache_vpi_return_annots_text(vpi_buf, src_sha1)
stdlib.free(vpi_buf)
stdlib.free(vpi_path)
return 1
# ============================================================
# _lookup_vpi_ret_annot - 查找缓存的方法返回注解字符串
# ============================================================
def _lookup_vpi_ret_annot(src_sha1: str, cls_name: str, method_name: str) -> str:
"""查找缓存的方法返回注解字符串,未找到返回 None"""
if _g_vpi_ret_keys is None or _g_vpi_ret_count == 0:
return None
if src_sha1 is None or method_name is None:
return None
sha1_len: t.CSizeT = string.strlen(src_sha1)
cl_len: t.CSizeT = 0
if cls_name is not None:
cl_len = string.strlen(cls_name)
me_len: t.CSizeT = string.strlen(method_name)
key_len: t.CSizeT = sha1_len + 1 + cl_len + 1 + me_len
key_buf: bytes = stdlib.malloc(key_len + 1)
if key_buf is None:
return None
if cls_name is not None:
viperlib.snprintf(key_buf, key_len + 1, "%s.%s.%s", src_sha1, cls_name, method_name)
else:
viperlib.snprintf(key_buf, key_len + 1, "%s.%s", src_sha1, method_name)
i: int
for i in range(_g_vpi_ret_count):
if string.strcmp(_g_vpi_ret_keys + i * VPI_KEY_LEN, key_buf) == 0:
stdlib.free(key_buf)
return _g_vpi_ret_strs + i * VPI_RET_LEN
stdlib.free(key_buf)
return None
# ============================================================
# _resolve_single_vpi_type_str - 解析单个类型段(无 CPtr/CVoid)为 LLVMType
#
# 纯字符串解析,不构造 AST 节点。
# ============================================================
def _resolve_single_vpi_type_str(pool: memhub.MemBuddy | t.CPtr,
seg: str) -> llvmlite.LLVMType | t.CPtr:
"""解析单个类型段为 LLVMType,无法识别返回 None"""
if seg is None or pool is None:
return None
i8_ty: llvmlite.LLVMType | t.CPtr = llvmlite.Int8(pool)
i8_ptr_ty: llvmlite.LLVMType | t.CPtr = llvmlite.Ptr(pool, i8_ty)
# 内置别名
if string.strcmp(seg, "str") == 0 or string.strcmp(seg, "bytes") == 0:
return i8_ptr_ty
if string.strcmp(seg, "int") == 0:
return llvmlite.Int32(pool)
if string.strcmp(seg, "bool") == 0:
return llvmlite.Int1(pool)
# t.XXX / c.XXX 内置类型映射
if string.strncmp(seg, "t.", 2) == 0 or string.strncmp(seg, "c.", 2) == 0:
attr_nm: str = seg + 2
mapped_ty: llvmlite.LLVMType | t.CPtr = HandlesType.map_t_type(pool, attr_nm)
if mapped_ty is not None:
return mapped_ty
return None
# 模块限定类名 "memhub.MemBuddy" → 取最后一段查 struct
last_dot: str = string.strrchr(seg, '.')
if last_dot is not None:
cls_nm2: str = last_dot + 1
mod_struct_ty: llvmlite.LLVMType | t.CPtr = HandlesStruct.get_struct_type(cls_nm2)
if mod_struct_ty is not None:
return mod_struct_ty
return None
# 裸类名 → 全局 struct 注册表
bare_struct_ty: llvmlite.LLVMType | t.CPtr = HandlesStruct.get_struct_type(seg)
if bare_struct_ty is not None:
return bare_struct_ty
# 裸 typedef 名(如 VOIDPTR)
mapped_bare: llvmlite.LLVMType | t.CPtr = HandlesType.map_t_type(pool, seg)
if mapped_bare is not None:
return mapped_bare
return None
# ============================================================
# _resolve_vpi_ret_annot_str - 直接解析 vpi 返回注解字符串为 LLVMType
#
# 背景:_parse_type_string 跨模块构造 ast.Name/ast.BinOp 在 TPV 中可能
# 退化为裸函数调用(__init__/vtable 未执行),节点 kind() 异常导致
# resolve 得到垃圾类型(如 Ptr(i8))。此处纯字符串解析避免该问题。
#
# 支持: "str"/"bytes" → i8*; "int" → i32; "bool" → i1;
# "t.CChar" 等 → map_t_type 映射;
# "X | t.CPtr" → Ptr(Struct X); "t.CVoid | t.CPtr" → i8*;
# "memhub.MemBuddy | t.CPtr" → Ptr(Struct MemBuddy)。
# ============================================================
def _resolve_vpi_ret_annot_str(pool: memhub.MemBuddy | t.CPtr,
ret_annot: str,
i32_ty: llvmlite.LLVMType | t.CPtr,
i8_ptr_ty: llvmlite.LLVMType | t.CPtr) -> llvmlite.LLVMType | t.CPtr:
"""直接解析 vpi 返回注解字符串为 LLVMType(不构造 AST 节点)"""
if ret_annot is None or pool is None:
return None
has_cptr: int = 0
has_cvoid: int = 0
first_ty: llvmlite.LLVMType | t.CPtr = None
buf_len: t.CSizeT = string.strlen(ret_annot)
pos: t.CSizeT = 0
while pos <= buf_len:
# 跳过前导空格
while pos < buf_len and ret_annot[pos] == ' ':
pos += 1
seg_start: t.CSizeT = pos
while pos < buf_len and ret_annot[pos] != '|':
pos += 1
seg_end: t.CSizeT = pos
# trim 尾随空格
while seg_end > seg_start and ret_annot[seg_end - 1] == ' ':
seg_end -= 1
if seg_end > seg_start:
seg_len2: t.CSizeT = seg_end - seg_start
seg_buf: bytes = stdlib.malloc(seg_len2 + 1)
if seg_buf is None:
break
string.strncpy(seg_buf, ret_annot + seg_start, seg_len2)
seg_buf[seg_len2] = '\0'
if string.strcmp(seg_buf, "t.CPtr") == 0 or string.strcmp(seg_buf, "CPtr") == 0:
has_cptr = 1
elif string.strcmp(seg_buf, "t.CVoid") == 0 or string.strcmp(seg_buf, "CVoid") == 0:
has_cvoid = 1
else:
seg_ty: llvmlite.LLVMType | t.CPtr = _resolve_single_vpi_type_str(pool, seg_buf)
if seg_ty is not None and first_ty is None:
first_ty = seg_ty
stdlib.free(seg_buf)
pos += 1 # 跳过 '|'
if has_cptr != 0:
if has_cvoid != 0:
return i8_ptr_ty
if first_ty is not None:
return llvmlite.Ptr(pool, first_ty)
return i8_ptr_ty
if first_ty is not None:
return first_ty
return None
# ============================================================
# VpiReader - .vpi AST 解析器
#
# 从 .vpi 存根文件读取函数签名和 CTypedef 定义,注入 declare
# 到当前模块。用于跨模块函数调用时补充缺失的函数声明。
#
# .vpi 文件由 StubGen.Converter._GeneratePyiFromAst 生成,
# 包含函数签名(无函数体)、类定义、变量声明、import 语句。
# VpiReader 解析 .vpi 为 AST,提取 FunctionDef 和 CTypedef
# AnnAssign,在当前模块创建对应的 declare 声明。
# ============================================================
# .vpi 读取缓冲区大小(256KB,与 PYI_BUF_SIZE 一致)
VPI_READ_BUF_SIZE: t.CSizeT = 262144
# SHA1 字符串长度(16 hex 字符)
SHA1_LEN: t.CSizeT = 16
# ============================================================
# _extract_sha1_from_path - 从 .vpi 文件路径提取 SHA1
#
# .vpi 文件名格式为 {sha1}.vpi,sha1 是 16 位十六进制字符。
# 从路径中找到最后一个路径分隔符(/ 或 \),取后续文件名的前
# 16 个字符作为 SHA1。分配新缓冲区存储 SHA1(调用者负责 free)。
#
# Args:
# vpi_path: .vpi 文件路径
#
# Returns:
# SHA1 字符串(stdlib.malloc 分配,需 stdlib.free),None 失败
# ============================================================
def _extract_sha1_from_path(vpi_path: str) -> str:
"""从 .vpi 文件路径提取 SHA1(文件名前16字符,malloc 分配)"""
if vpi_path is None:
return None
# 查找最后一个路径分隔符 / 或 \
total_len: t.CSizeT = string.strlen(vpi_path)
if total_len < SHA1_LEN + 4: # 至少需要 "xxxx.vpi" 的长度
return None
# 从后向前查找最后一个 / 或 \
sep_pos: t.CSizeT = total_len
while sep_pos > 0:
sep_pos -= 1
ch: t.CChar = vpi_path[sep_pos]
if ch == '/' or ch == '\\':
sep_pos += 1
break
if sep_pos + SHA1_LEN > total_len:
return None
# 分配缓冲区并复制 SHA1(16 字符 + null 终止符)
sha1_buf: str = stdlib.malloc(SHA1_LEN + 1)
if sha1_buf is None:
return None
string.strncpy(sha1_buf, vpi_path + sep_pos, SHA1_LEN)
sha1_buf[SHA1_LEN] = '\0'
return sha1_buf
# ============================================================
# _stub_add_imported - 追加模块名到空格分隔列表(stub 上下文收集用)
#
# 与 HandlesImports.add_imported_module 逻辑一致,但内联在本模块,
# 避免 VpiReader → HandlesImports → StubMerger → VpiReader 循环依赖。
# ============================================================
def _stub_add_imported(pool: memhub.MemBuddy | t.CPtr,
cur: str, name: str) -> str:
"""追加模块名到空格分隔列表(pool 分配),返回新列表"""
if name is None:
return cur
if cur is None:
nlen: t.CSizeT = string.strlen(name)
buf: str = pool.alloc(nlen + 1)
if buf is not None:
string.strcpy(buf, name)
return buf
old_len: t.CSizeT = string.strlen(cur)
name_len: t.CSizeT = string.strlen(name)
buf2: str = pool.alloc(old_len + 1 + name_len + 1)
if buf2 is not None:
string.strcpy(buf2, cur)
buf2[old_len] = ' '
string.strcpy(buf2 + old_len + 1, name)
return buf2
# ============================================================
# _stub_add_from - 追加 from-import 映射("local:mod[:orig]" 空格分隔)
# ============================================================
def _stub_add_from(pool: memhub.MemBuddy | t.CPtr,
cur: str, local: str, mod: str, orig: str) -> str:
"""追加 from-import 映射(pool 分配),返回新列表"""
if local is None or mod is None:
return cur
entry: str = pool.alloc(256)
if entry is None:
return cur
if orig is not None and string.strcmp(orig, local) != 0:
viperlib.snprintf(entry, 256, "%s:%s:%s", local, mod, orig)
else:
viperlib.snprintf(entry, 256, "%s:%s", local, mod)
if cur is None:
return entry
old_len: t.CSizeT = string.strlen(cur)
entry_len: t.CSizeT = string.strlen(entry)
buf: str = pool.alloc(old_len + 1 + entry_len + 1)
if buf is not None:
string.strcpy(buf, cur)
buf[old_len] = ' '
string.strcpy(buf + old_len + 1, entry)
return buf
# ============================================================
# ParseVpiIntoModule - 解析 .vpi 文件,将函数声明注入到当前模块
#
# 读取 .vpi 文件 → AST 解析 → 遍历顶层节点 → 创建 declare 声明。
# 对每个 FunctionDef:
# 1. 提取函数名、参数、返回类型
# 2. 检测 CExport/CExtern/State 标记(决定是否用裸名)
# 3. 用 find_func_in_module 去重(已存在则跳过)
# 4. 用 create_declare + add_param 创建 declare 声明
# 5. 注册到函数表(用裸名,非 mangled_name)
# 对 AnnAssign with t.CTypedef:调用 register_ctypedef 注册
# ClassDef/Import/ImportFrom:跳过
#
# Args:
# pool: 内存池
# mod: 当前 LLVM 模块
# vpi_path: .vpi 文件路径
# trans: 翻译器对象
#
# Returns:
# 1=成功,0=失败
# ============================================================
def ParseVpiIntoModule(pool: memhub.MemBuddy | t.CPtr,
mod: llvmlite.LLVMModule | t.CPtr,
vpi_path: str,
trans: HT.Translator | t.CPtr) -> int:
"""解析 .vpi 文件,将函数声明注入到当前模块
解析 stub 期间临时将错误归因切换到被读 stub 文件(.vpi):
注解解析失败时 fatal_error 应指向 stub 及其行号(真实出错位置),
而非当前编译模块(错误归因修复)。
"""
if pool is None or mod is None or vpi_path is None or trans is None:
return 0
saved_cur: str = HandlesType.get_current_file()
HandlesType.set_current_file(vpi_path)
ret: int = _ParseVpiIntoModuleImpl(pool, mod, vpi_path, trans)
HandlesType.set_current_file(saved_cur)
return ret
def _ParseVpiIntoModuleImpl(pool: memhub.MemBuddy | t.CPtr,
mod: llvmlite.LLVMModule | t.CPtr,
vpi_path: str,
trans: HT.Translator | t.CPtr) -> int:
"""解析 .vpi 文件,将函数声明注入到当前模块(ParseVpiIntoModule 实际实现)"""
if pool is None or mod is None or vpi_path is None or trans is None:
return 0
# 1. 读取 .vpi 文件
f: fileio.File | t.CPtr = fileio.File(vpi_path, fileio.MODE.R)
if f.closed:
return 0
vpi_buf: bytes = stdlib.malloc(VPI_READ_BUF_SIZE)
if vpi_buf is None:
f.close()
return 0
bytes_read: LONG = f.read_all(vpi_buf, VPI_READ_BUF_SIZE)
f.close()
if bytes_read <= 0:
stdlib.free(vpi_buf)
return 0
if bytes_read < VPI_READ_BUF_SIZE:
vpi_buf[bytes_read] = 0
else:
vpi_buf[VPI_READ_BUF_SIZE - 1] = 0
# 2. 解析为 AST
lx: ast.Lexer | t.CPtr = ast.new_lexer(pool)
if lx is None:
stdlib.free(vpi_buf)
return 0
ast._lexer_init(lx, vpi_buf, pool)
tokens: ast.Token | t.CPtr = ast.tokenize(lx)
tree: ast.AST | t.CPtr = ast.parse_tokens(pool, tokens)
# vpi_buf 延迟释放:AST 字符串字段可能引用 vpi_buf 内部(token 零拷贝),
# 提前 free 会导致 Name.id/Attribute.attr 等悬空 → 后续解析返回错误类型。
# 在 ParseVpiIntoModule 末尾统一释放。
if tree is None:
stdlib.free(vpi_buf)
return 0
# 提取源模块 SHA1(用于函数名混淆,在 AST 解析成功后提取避免提前返回时泄漏)
src_sha1: str = _extract_sha1_from_path(vpi_path)
# 从 vpi_buf 文本直接提取方法返回注解字符串缓存到 stdlib
# (不依赖 AST tree:tree 深层节点在后续遍历处理时可能被覆盖损坏,
# 但 vpi_buf 原始文本在 ParseVpiIntoModule 末尾才释放,始终有效)
_cache_vpi_return_annots_text(vpi_buf, src_sha1)
# 3. 遍历 AST 顶层节点
ch: list[ast.AST | t.CPtr] | t.CPtr = tree.children
if ch is None:
if src_sha1 is not None:
stdlib.free(src_sha1)
return 1
n: t.CSizeT = ch.__len__()
# 收集 stub 内被读模块自己的 import 语句,构建独立的注解解析上下文。
# .vpi stub 顶部含被读模块的 import(Converter 已写入,如 string.vp 的
# "import t, c"),用它替代当前模块的 imported_modules/from_imports 去
# 解析注解——根治跨上下文误报(如 t.CArray[str] 需要 't' 已导入)。
stub_imported_modules: str = None
stub_from_imports: str = None
si: t.CSizeT
for si in range(n):
schild: ast.AST | t.CPtr = ch.get(si)
if schild is None:
continue
sk: int = schild.kind()
if sk == ast.ASTKind.Import:
simp: ast.Import | t.CPtr = (ast.Import | t.CPtr)(schild)
if simp.names is not None:
sn_n: t.CSizeT = simp.names.__len__()
sni: t.CSizeT
for sni in range(sn_n):
sal: ast.Alias | t.CPtr = (ast.Alias | t.CPtr)(simp.names.get(sni))
if sal is not None and sal.name is not None:
stub_imported_modules = _stub_add_imported(
pool, stub_imported_modules, sal.name)
if sal.asname is not None:
stub_imported_modules = _stub_add_imported(
pool, stub_imported_modules, sal.asname)
elif sk == ast.ASTKind.ImportFrom:
sifr: ast.ImportFrom | t.CPtr = (ast.ImportFrom | t.CPtr)(schild)
sifr_mod: str = sifr.module
if sifr_mod is not None:
stub_imported_modules = _stub_add_imported(
pool, stub_imported_modules, sifr_mod)
if sifr.names is not None:
sifr_nn: t.CSizeT = sifr.names.__len__()
sifri: t.CSizeT
for sifri in range(sifr_nn):
sial: ast.Alias | t.CPtr = (ast.Alias | t.CPtr)(sifr.names.get(sifri))
if sial is not None and sial.name is not None:
# '*' star import 不建 from 映射(lookup 有 star 回退)
if string.strcmp(sial.name, "*") != 0:
slocal: str = sial.name
sorig: str = None
if sial.asname is not None:
slocal = sial.asname
sorig = sial.name
stub_from_imports = _stub_add_from(
pool, stub_from_imports, slocal, sifr_mod, sorig)
# 用 stub 自己的 imports 替代当前模块的(stub 注解解析上下文正确性)
imported_modules: str = trans._imported_modules
from_imports: str = trans._from_imports
if stub_imported_modules is not None:
imported_modules = stub_imported_modules
if stub_from_imports is not None:
from_imports = stub_from_imports
i32_ty: llvmlite.LLVMType | t.CPtr = llvmlite.Int32(pool)
i8_ty: llvmlite.LLVMType | t.CPtr = llvmlite.Int8(pool)
i8_ptr_ty: llvmlite.LLVMType | t.CPtr = llvmlite.Ptr(pool, i8_ty)
# 函数表信息
funcs_ptr: HandlesExprCall.FuncEntry | t.CPtr = trans._funcs
max_funcs: int = 256
i: t.CSizeT
for i in range(n):
child: ast.AST | t.CPtr = ch.get(i)
if child is None:
continue
kd: int = child.kind()
# FunctionDef → 提取函数名、参数、返回类型 → 注入 declare
if kd == ast.ASTKind.FunctionDef:
fd: ast.FunctionDef | t.CPtr = (ast.FunctionDef | t.CPtr)(child)
if fd is None or fd.name is None:
continue
_inject_function_declare(pool, mod, fd, trans, src_sha1,
imported_modules, from_imports,
i32_ty, i8_ptr_ty,
funcs_ptr, max_funcs)
continue
# AnnAssign with t.CTypedef → 注册 CTypedef
if kd == ast.ASTKind.AnnAssign:
aa: ast.AnnAssign | t.CPtr = (ast.AnnAssign | t.CPtr)(child)
if aa is None:
continue
if HandlesType._is_ctypedef_annotation(aa.annotation) != 0:
_register_ctypedef_from_annassign(pool, aa)
continue
# ClassDef → 递归注入类内方法(如 MemBuddy.__init__/alloc)
# 方法 declare 使用 {sha1}.{ClassName}.{method} 混淆名,
# 参数类型从 vpi 注解解析(避免调用时 _ensure_method_declare 用实参类型
# 创建错误 declare,如 size: t.CSizeT 被写成 i32 而非 i64)
if kd == ast.ASTKind.ClassDef:
class_cd: ast.ClassDef | t.CPtr = (ast.ClassDef | t.CPtr)(child)
if class_cd is None or class_cd.name is None:
continue
cls_name_v: str = class_cd.name
cls_children: list[ast.AST | t.CPtr] | t.CPtr = class_cd.children
if cls_children is not None:
cn2: t.CSizeT = cls_children.__len__()
for ci2 in range(cn2):
m_child: ast.AST | t.CPtr = cls_children.get(ci2)
if m_child is None:
continue
if m_child.kind() != ast.ASTKind.FunctionDef:
continue
m_fd: ast.FunctionDef | t.CPtr = (ast.FunctionDef | t.CPtr)(m_child)
if m_fd is None or m_fd.name is None:
continue
_inject_method_declare(pool, mod, m_fd, cls_name_v, trans,
src_sha1, imported_modules, from_imports,
i32_ty, i8_ptr_ty, funcs_ptr, max_funcs)
continue
# Import/ImportFrom → 跳过
# (import 不需要处理)
# 释放 SHA1 缓冲区(_extract_sha1_from_path 用 stdlib.malloc 分配)
if src_sha1 is not None:
stdlib.free(src_sha1)
# vpi_buf 延迟到全部 tree 处理完成后释放(见解析处注释)
stdlib.free(vpi_buf)
return 1
# ============================================================
# _inject_function_declare - 为单个 FunctionDef 创建 declare 声明
#
# 检测 CExport/CExtern/State 标记,计算 mangled name,
# 用 find_func_in_module 去重,创建 declare 并添加参数,
# 注册到函数表。
# ============================================================
def _inject_function_declare(pool: memhub.MemBuddy | t.CPtr,
mod: llvmlite.LLVMModule | t.CPtr,
fd: ast.FunctionDef | t.CPtr,
trans: HT.Translator | t.CPtr,
src_sha1: str,
imported_modules: str,
from_imports: str,
i32_ty: llvmlite.LLVMType | t.CPtr,
i8_ptr_ty: llvmlite.LLVMType | t.CPtr,
funcs_ptr: HandlesExprCall.FuncEntry | t.CPtr,
max_funcs: int) -> None:
"""为 FunctionDef 创建 declare 声明并注入到模块"""
if fd.name is None:
return
# 检测 CExtern/State/CExport 标记(从返回类型注解)
has_extern: int = 0
has_state: int = 0
has_export: int = 0
if fd.returns is not None:
has_extern = HandlesType.has_decorator_marker(fd.returns, "CExtern")
has_state = HandlesType.has_decorator_marker(fd.returns, "State")
has_export = HandlesType.has_decorator_marker(fd.returns, "CExport")
# 计算 mangled name
# CExport/CExtern/State/main 函数用裸名,其他函数用源模块 SHA1 前缀
use_bare_name: int = 0
if has_extern != 0 or has_state != 0 or has_export != 0 or string.strcmp(fd.name, "main") == 0:
mangled_name: str = fd.name
use_bare_name = 1
else:
mangled_name = HandlesFunctions._mangle_name_with_sha1(src_sha1, fd.name)
# 去重:如果函数已存在则跳过
existing: llvmlite.Function | t.CPtr = HandlesExprCall.find_func_in_module(mod, mangled_name)
if existing is not None:
return
# 也检查裸名(非裸名模式时,CExport 函数可能已通过其他途径注册)
if use_bare_name == 0:
existing_bare: llvmlite.Function | t.CPtr = HandlesExprCall.find_func_in_module(mod, fd.name)
if existing_bare is not None:
return
# 推断返回类型
# 优先用缓存的返回注解字符串(tree 解析后立即提取,节点未损坏)
# 直接字符串解析(绕开 _parse_type_string 跨模块 AST 构造在 TPV 中的
# __init__/vtable 未执行问题,节点 kind() 异常导致 resolve 得垃圾类型)
ret_ty: llvmlite.LLVMType | t.CPtr = None
ret_annot_fd: str = _lookup_vpi_ret_annot(src_sha1, None, fd.name)
if ret_annot_fd is not None:
ret_ty = _resolve_vpi_ret_annot_str(pool, ret_annot_fd, i32_ty, i8_ptr_ty)
if ret_ty is None and fd.returns is not None:
ret_ty = HandlesType.resolve_annotation_type(
pool, fd.returns, imported_modules, from_imports, trans)
# 如果返回类型是纯 t.State(无实际类型),用 Void
if ret_ty is None and fd.returns is not None:
if has_state != 0:
ret_ty = llvmlite.Void(pool)
# 默认返回 i32
if ret_ty is None:
ret_ty = i32_ty
# 创建 declare
func: llvmlite.Function | t.CPtr = llvmlite.create_declare(
pool, mod, mangled_name, ret_ty)
if func is None:
return
# 提取参数并添加到 declare
param_count: int = 0
if fd.args is not None:
args_node: ast.Arguments | t.CPtr = (ast.Arguments | t.CPtr)(fd.args)
if args_node.args is not None:
args_list: list[ast.AST | t.CPtr] | t.CPtr = args_node.args
param_count = args_list.__len__()
pi: t.CSizeT
for pi in range(param_count):
ag: ast.Arg | t.CPtr = (ast.Arg | t.CPtr)(args_list.get(pi))
if ag is None:
continue
pname: str = ag.arg
if pname is None:
pname = "arg"
# 解析参数类型
param_ty: llvmlite.LLVMType | t.CPtr = i8_ptr_ty
if ag.annotation is not None:
resolved_ty: llvmlite.LLVMType | t.CPtr = HandlesType.resolve_annotation_type(
pool, ag.annotation, imported_modules, from_imports, trans)
if resolved_ty is not None:
# 结构体按值参数 → i64:跨模块结构体在本模块是 opaque
# (未定义字段),LLVM 不允许 opaque struct 作为函数参数
# (llc: invalid type for function argument)。Windows ABI
# 小结构体(≤8 字节,如 COORD)按整数寄存器传,映射 i64 兼容。
match resolved_ty:
case llvmlite.LLVMType.Struct(_, _, _):
param_ty = llvmlite.Int64(pool)
case _:
param_ty = resolved_ty
llvmlite.add_param(pool, func, param_ty, pname)
# *args 变长参数:标记 IsVarArg=1,使调用侧走 C ABI 参数提升
# (i8/i16 → i32,float → double),避免 %d/%x 等格式因参数按 i64
# 传递而读错(第二个 %x 读到第一个 i64 槽的高 4 字节)。
if args_node.vararg is not None:
func.IsVarArg = 1
# 注册到函数表(用裸名 fd.name,不是 mangled_name)
if funcs_ptr is not None:
cur_count: int = trans._func_count
if HandlesExprCall.add_func_to_table(funcs_ptr, cur_count, fd.name, func, max_funcs,
None, 0, param_count) == 0:
trans._func_count = cur_count + 1
# ============================================================
# _inject_method_declare - 为 ClassDef 内的方法创建 declare
#
# 方法混淆名: {src_sha1}.{ClassName}.{method}
# 参数类型从 vpi 注解解析(self 无注解时默认为 i8*)。
# 注入正确签名避免调用时 _ensure_method_declare 用实参类型
# 创建错误 declare(如 size: t.CSizeT 被写成 i32 而非 i64)。
# ============================================================
def _inject_method_declare(pool: memhub.MemBuddy | t.CPtr,
mod: llvmlite.LLVMModule | t.CPtr,
m_fd: ast.FunctionDef | t.CPtr,
cls_name: str,
trans: HT.Translator | t.CPtr,
src_sha1: str,
imported_modules: str,
from_imports: str,
i32_ty: llvmlite.LLVMType | t.CPtr,
i8_ptr_ty: llvmlite.LLVMType | t.CPtr,
funcs_ptr: HandlesExprCall.FuncEntry | t.CPtr,
max_funcs: int) -> None:
"""为 ClassDef 内方法创建 declare 声明并注入到模块"""
if m_fd is None or m_fd.name is None or cls_name is None:
return
if src_sha1 is None:
return
# 构造方法查找名: ClassName.method
lookup_name_len: t.CSizeT = string.strlen(cls_name) + 1 + string.strlen(m_fd.name) + 1
lookup_name: str = pool.alloc(lookup_name_len)
if lookup_name is None:
return
viperlib.snprintf(lookup_name, lookup_name_len, "%s.%s", cls_name, m_fd.name)
# mangled name: {src_sha1}.{ClassName}.{method}
mangled_len: t.CSizeT = string.strlen(src_sha1) + 1 + lookup_name_len
mangled_name: str = pool.alloc(mangled_len)
if mangled_name is None:
return
viperlib.snprintf(mangled_name, mangled_len, "%s.%s", src_sha1, lookup_name)
# 去重:如果方法已存在则跳过
existing: llvmlite.Function | t.CPtr = HandlesExprCall.find_func_in_module(mod, mangled_name)
if existing is not None:
return
# 推断返回类型
# 优先用缓存的返回注解字符串(tree 解析后立即提取,节点未损坏),
# 用 _parse_type_string 重新解析成新 AST(立即使用不被覆盖)。
# 直接解析 m_fd.returns 可能因 vpi tree 深层节点被覆盖而得到垃圾类型。
ret_ty: llvmlite.LLVMType | t.CPtr = None
ret_annot: str = _lookup_vpi_ret_annot(src_sha1, cls_name, m_fd.name)
if ret_annot is not None:
ret_ty = _resolve_vpi_ret_annot_str(pool, ret_annot, i32_ty, i8_ptr_ty)
if ret_ty is None and m_fd.returns is not None:
ret_ty = HandlesType.resolve_annotation_type(
pool, m_fd.returns, imported_modules, from_imports, trans)
# __new__ 无返回注解时推断为类指针(__new__ 返回 self 指针,默认 i32 会截断指针)
if ret_ty is None and string.strcmp(m_fd.name, "__new__") == 0:
cls_ty: llvmlite.LLVMType | t.CPtr = HandlesStruct.get_struct_type(cls_name)
if cls_ty is not None:
ret_ty = llvmlite.Ptr(pool, cls_ty)
if ret_ty is None:
ret_ty = i32_ty
# 创建 declare
func: llvmlite.Function | t.CPtr = llvmlite.create_declare(
pool, mod, mangled_name, ret_ty)
if func is None:
return
# 提取参数并添加到 declare
param_count: int = 0
if m_fd.args is not None:
args_node: ast.Arguments | t.CPtr = (ast.Arguments | t.CPtr)(m_fd.args)
if args_node.args is not None:
args_list: list[ast.AST | t.CPtr] | t.CPtr = args_node.args
param_count = args_list.__len__()
pi: t.CSizeT
for pi in range(param_count):
ag: ast.Arg | t.CPtr = (ast.Arg | t.CPtr)(args_list.get(pi))
if ag is None:
continue
pname: str = ag.arg
if pname is None:
pname = "arg"
# 解析参数类型
param_ty: llvmlite.LLVMType | t.CPtr = i8_ptr_ty
if ag.annotation is not None:
resolved_ty: llvmlite.LLVMType | t.CPtr = HandlesType.resolve_annotation_type(
pool, ag.annotation, imported_modules, from_imports, trans)
if resolved_ty is not None:
# 结构体按值参数 → i64:跨模块结构体在本模块是 opaque
# (未定义字段),LLVM 不允许 opaque struct 作为函数参数
# (llc: invalid type for function argument)。Windows ABI
# 小结构体(≤8 字节,如 COORD)按整数寄存器传,映射 i64 兼容。
match resolved_ty:
case llvmlite.LLVMType.Struct(_, _, _):
param_ty = llvmlite.Int64(pool)
case _:
param_ty = resolved_ty
llvmlite.add_param(pool, func, param_ty, pname)
# 注册到函数表(用裸名 ClassName.method,与调用方 lookup_name 一致)
if funcs_ptr is not None:
cur_count: int = trans._func_count
if HandlesExprCall.add_func_to_table(funcs_ptr, cur_count, lookup_name, func, max_funcs,
None, 0, param_count) == 0:
trans._func_count = cur_count + 1
# ============================================================
# _register_ctypedef_from_annassign - 从 AnnAssign 节点注册 CTypedef
#
# 提取 NAME: t.CTypedef = Type 中的名称和类型表达式字符串,
# 调用 register_ctypedef 注册到动态 CTypedef 表。
# ============================================================
def _register_ctypedef_from_annassign(pool: memhub.MemBuddy | t.CPtr,
aa: ast.AnnAssign | t.CPtr) -> None:
"""从 AnnAssign 节点提取并注册 CTypedef"""
if aa is None or pool is None:
return
# 提取目标名称(必须是 Name 节点)
if aa.target is None or aa.target.kind() != ast.ASTKind.Name:
return
tgt: ast.Name | t.CPtr = (ast.Name | t.CPtr)(aa.target)
if tgt.id is None:
return
# 提取类型表达式字符串(从 value AST 节点)
if aa.value is None:
return
type_str: str = HandlesType._get_type_name_from_annotation(pool, aa.value)
if type_str is None:
return
# 注册 CTypedef
HandlesType.register_ctypedef(pool, tgt.id, type_str)