869 lines
39 KiB
Plaintext
869 lines
39 KiB
Plaintext
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*）。
|
||
# 因此在 tree 解析后立即把每个方法的返回注解转成字符串缓存到 stdlib
|
||
# （持久，不受 pool 状态影响），_inject_method_declare 用 _parse_type_string
|
||
# 重新解析缓存字符串（新 AST 节点，立即使用不被覆盖）。
|
||
# ============================================================
|
||
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
|
||
|
||
|
||
# ============================================================
|
||
# _cache_vpi_return_annots - tree 解析后立即缓存方法的返回注解字符串
|
||
# ============================================================
|
||
def _cache_vpi_return_annots(pool: memhub.MemBuddy | t.CPtr,
|
||
tree: ast.AST | t.CPtr,
|
||
src_sha1: str) -> int:
|
||
"""缓存 vpi tree 中所有 FunctionDef 的返回注解字符串（stdlib 持久）"""
|
||
global _g_vpi_ret_keys, _g_vpi_ret_strs, _g_vpi_ret_count
|
||
if tree 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)
|
||
sha1_len: t.CSizeT = string.strlen(src_sha1)
|
||
ch: list[ast.AST | t.CPtr] | t.CPtr = tree.children
|
||
if ch is None:
|
||
return 0
|
||
n: t.CSizeT = ch.__len__()
|
||
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()
|
||
if kd == ast.ASTKind.ClassDef:
|
||
cd: ast.ClassDef | t.CPtr = (ast.ClassDef | t.CPtr)(child)
|
||
if cd is None or cd.name is None:
|
||
continue
|
||
mch: list[ast.AST | t.CPtr] | t.CPtr = cd.children
|
||
if mch is None:
|
||
continue
|
||
mn: t.CSizeT = mch.__len__()
|
||
mi: t.CSizeT
|
||
for mi in range(mn):
|
||
m_child: ast.AST | t.CPtr = mch.get(mi)
|
||
if m_child is None or 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
|
||
_store_vpi_ret_entry(src_sha1, sha1_len, cd.name, m_fd.name, m_fd.returns)
|
||
elif kd == ast.ASTKind.FunctionDef:
|
||
fd: ast.FunctionDef | t.CPtr = (ast.FunctionDef | t.CPtr)(child)
|
||
if fd is None or fd.name is None:
|
||
continue
|
||
_store_vpi_ret_entry(src_sha1, sha1_len, None, fd.name, fd.returns)
|
||
return 1
|
||
|
||
|
||
def _store_vpi_ret_entry(src_sha1: str, sha1_len: t.CSizeT,
|
||
cls_name: str, method_name: str,
|
||
returns: ast.AST | t.CPtr) -> None:
|
||
"""存储单个方法的返回注解字符串到缓存"""
|
||
global _g_vpi_ret_count
|
||
if method_name is None:
|
||
return
|
||
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
|
||
if key_len >= VPI_KEY_LEN:
|
||
return
|
||
key_buf: bytes = stdlib.malloc(key_len + 1)
|
||
if key_buf is None:
|
||
return
|
||
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)
|
||
ret_str: str = None
|
||
if returns is not None:
|
||
ret_str = HandlesType.annotation_to_str(returns)
|
||
if ret_str is not None and string.strlen(ret_str) > 0:
|
||
if string.strlen(ret_str) < VPI_RET_LEN:
|
||
if _g_vpi_ret_count < VPI_RET_MAX:
|
||
string.strcpy(_g_vpi_ret_keys + _g_vpi_ret_count * VPI_KEY_LEN, key_buf)
|
||
string.strcpy(_g_vpi_ret_strs + _g_vpi_ret_count * VPI_RET_LEN, ret_str)
|
||
_g_vpi_ret_count += 1
|
||
stdlib.free(key_buf)
|
||
|
||
|
||
# ============================================================
|
||
# _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
|
||
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)
|
||
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 and _g_vpi_ret_count < VPI_RET_MAX:
|
||
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
|
||
|
||
|
||
# ============================================================
|
||
# _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
|
||
|
||
|
||
# ============================================================
|
||
# 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
|
||
|
||
# 获取当前翻译器的导入信息（供 resolve_annotation_type 使用）
|
||
imported_modules: str = trans._imported_modules
|
||
from_imports: str = trans._from_imports
|
||
|
||
# 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__()
|
||
|
||
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)
|