Files
ViperC/App/lib/StubGen/VpiReader.vp
T

481 lines
21 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 stdio
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
# ============================================================
# _ddebug_exit - DDEBUG 拦截退出辅助函数
#
# 刷新 stderr 缓冲区并通过 VLogger 输出可见消息后退出。
# 解决 sys.exit(1) 不刷新缓冲区导致 DDEBUG 消息丢失的问题。
# ============================================================
def _ddebug_exit() -> None:
stdio.fflush(stdio.stderr)
VLogger.error("DDEBUG intercept triggered - check stderr for details")
# ============================================================
# 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'
# DDEBUG: 检测从路径提取的 SHA1 是否带尾随点(>=17 字符),发现问题立即终止
_ext_sl: t.CSizeT = string.strlen(sha1_buf)
if _ext_sl >= 17:
stdio.printf("[DDEBUG] _extract_sha1_from_path too long: len=%d sha1='%s' path='%s'\n", _ext_sl, sha1_buf, vpi_path)
_ddebug_exit()
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 文件,将函数声明注入到当前模块"""
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)
stdlib.free(vpi_buf)
if tree is None:
return 0
# 提取源模块 SHA1(用于函数名混淆,在 AST 解析成功后提取避免提前返回时泄漏)
src_sha1: str = _extract_sha1_from_path(vpi_path)
# 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)
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)
# DDEBUG: VpiReader 是注入模块级函数 declare 的路径(无拦截,最可疑双点源头)
# 检测 src_sha1 带尾随点 / fd.name 带前导点 / mangled_name 含双点,发现立即终止
if mangled_name is not None and src_sha1 is not None and fd.name is not None:
_vpi_sl: t.CSizeT = string.strlen(src_sha1)
if _vpi_sl >= 17:
stdio.printf("[DDEBUG] VpiReader src_sha1 too long: len=%d sha1='%s' func='%s'\n", _vpi_sl, src_sha1, fd.name)
_ddebug_exit()
_vpi_fn_len: t.CSizeT = string.strlen(fd.name)
if _vpi_fn_len > 0 and fd.name[0] == '.':
stdio.printf("[DDEBUG] VpiReader fd.name starts with dot: name='%s' sha1='%s'\n", fd.name, src_sha1)
_ddebug_exit()
_vpi_mi: int = 0
_vpi_ml: t.CSizeT = string.strlen(mangled_name)
while _vpi_mi < _vpi_ml - 1:
if mangled_name[_vpi_mi] == 46 and mangled_name[_vpi_mi + 1] == 46:
stdio.printf("[DDEBUG] VpiReader mangled double-dot: sha1='%s' func='%s' mangled='%s'\n", src_sha1, fd.name, mangled_name)
_ddebug_exit()
_vpi_mi += 1
# 去重:如果函数已存在则跳过
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
# 推断返回类型
ret_ty: llvmlite.LLVMType | t.CPtr = None
if 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:
param_ty = resolved_ty
llvmlite.add_param(pool, func, param_ty, pname)
# 注册到函数表(用裸名 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
# 推断返回类型
ret_ty: llvmlite.LLVMType | t.CPtr = None
if 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:
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)