TermiNexus
53c72ff57a
main.vp IR 显示 intr.gdt.init() 解析成 @3b3cc38844ed7ad9.init（serial.vp）、
intr.idt.init() 解析成 @c418c6220a9a9d2a.init（su8250.vp，还带伪造 i16 实参）
——intr 包限定的跨模块调用因 is_module_imported('intr.gdt')=0 跳过校验，
find_func_in_module 后缀匹配劫持最先注册的同名 init → GDT/IDT 从未初始化，
CPU 一直用 OVMF 的 IDT（int 0x20 → OVMF #GP 0xEB6810C，定时中断永不到达）。
新增 ImportsGen.is_module_imported_or_known：已导入 或 全局已登记且任一祖先
包已导入（import intr → intr.gdt 可访问；this.this() 未 import this 仍 fast-fail）。
替换 CallGenMixin 模块限定解析 5 处门控（构造单层/多层、found_func 校验、
跨模块调用 import 检查）。
829 lines
41 KiB
Plaintext
829 lines
41 KiB
Plaintext
import lib.core.Gen.Generator as Generator
|
||
import t, c
|
||
from stdint import *
|
||
import ast
|
||
import memhub
|
||
import string
|
||
import stdlib
|
||
import viperlib
|
||
import stdio
|
||
import sys
|
||
import w32.fileio as fileio
|
||
import lib.Stub.StubMerger as StubMerger
|
||
import lib.core.CompilerFactory as CompilerFactory
|
||
import lib.core.Handles.HandlesTranslator as HT
|
||
import lib.Stub.StubMerger.StubMergerStore as StubMergerStore
|
||
|
||
|
||
# ============================================================
|
||
# ImportsGenMixin - 导入处理 Mixin（对象化 P6-D，wiki/14-oo-refactor.md）
|
||
#
|
||
# 从 HandlesImports.vp 迁移：
|
||
# 全局状态（_g_reexport_sha1s/_g_reexport_funcs/_g_reexport_srcs/
|
||
# _g_reexport_count/_g_current_file）→ 本类字段
|
||
# 模块级函数（add_imported_module/is_module_imported/...）→ 本类方法
|
||
#
|
||
# 迁移原则（与 P1-P5 Mixin 一致，行为零变化）：
|
||
# 1. 方法保留原函数签名（含 pool 首参），仅函数体内的全局变量改为 self.字段
|
||
# 2. 本模块内部函数调用改为 self.方法；跨模块调用（StubMerger 等）保留模块前缀
|
||
# 3. HandlesImports.vp 原函数保留为"委托"：本地模块级单例 _g_gen
|
||
# （BuildPipeline/main 注入）cast 成 Generator 类型后调本类方法
|
||
#
|
||
# @t.NoVTable：静态绑定（方法沿 MRO 直接调用 + this 调整，无 vtable 开销）。
|
||
# ============================================================
|
||
|
||
|
||
# ============================================================
|
||
# 常量（与 HandlesImports 保持一致）
|
||
# ============================================================
|
||
MAX_REEXPORT: t.CDefine = 512
|
||
|
||
|
||
# ============================================================
|
||
# ImportsGenMixin - 导入处理
|
||
# ============================================================
|
||
@t.NoVTable
|
||
class ImportsGenMixin:
|
||
"""导入处理 Mixin（P6 方法化）"""
|
||
|
||
# --- 全局 re-export 映射表（原 _g_reexport_sha1s/_g_reexport_funcs/
|
||
# _g_reexport_srcs/_g_reexport_count） ---
|
||
Gen: Generator.Generator | t.CPtr
|
||
ReexportSha1s: bytes
|
||
ReexportFuncs: bytes
|
||
ReexportSrcs: bytes
|
||
ReexportCount: int
|
||
|
||
# 注意：不声明 CurrentFile 字段——TypeGenMixin 已声明（Generator 展平块
|
||
# 只有一份，跨父类不去重规则下重复声明会导致字段错位）。本 Mixin 的
|
||
# set_current_file 与 CurrentFile 读取统一经 HandlesType 委托（延迟 import
|
||
# 规避 HandlesType↔HandlesImports 循环依赖）。
|
||
|
||
# ============================================================
|
||
# 添加已导入模块名
|
||
# ============================================================
|
||
def add_imported_module(self, pool: memhub.MemBuddy | t.CPtr,
|
||
imported_modules: str,
|
||
name: str) -> str:
|
||
"""记录已导入的模块名，返回新的 imported_modules 字符串指针"""
|
||
if name is None:
|
||
return imported_modules
|
||
if imported_modules 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
|
||
return None
|
||
else:
|
||
old_len: t.CSizeT = string.strlen(imported_modules)
|
||
name_len: t.CSizeT = string.strlen(name)
|
||
new_len: t.CSizeT = old_len + 1 + name_len
|
||
buf2: str = pool.alloc(new_len + 1)
|
||
if buf2 is not None:
|
||
string.strcpy(buf2, imported_modules)
|
||
buf2[old_len] = ' '
|
||
string.strcpy(buf2 + old_len + 1, name)
|
||
return buf2
|
||
return imported_modules
|
||
|
||
# ============================================================
|
||
# 检查模块是否已导入
|
||
# ============================================================
|
||
def is_module_imported(self, imported_modules: str, name: str) -> int:
|
||
"""检查模块是否已导入（词边界精确匹配，避免子串误匹配）"""
|
||
if name is None or imported_modules is None:
|
||
return 0
|
||
name_len: t.CSizeT = string.strlen(name)
|
||
cur: str = imported_modules
|
||
total_len: t.CSizeT = string.strlen(imported_modules)
|
||
ci: t.CSizeT = 0
|
||
while ci < total_len:
|
||
while ci < total_len and cur[ci] == ' ':
|
||
ci += 1
|
||
if ci >= total_len:
|
||
break
|
||
word_start: t.CSizeT = ci
|
||
while ci < total_len and cur[ci] != ' ':
|
||
ci += 1
|
||
word_len: t.CSizeT = ci - word_start
|
||
if word_len == name_len:
|
||
match: int = 1
|
||
for ei in range(name_len):
|
||
if cur[word_start + ei] != name[ei]:
|
||
match = 0
|
||
break
|
||
if match == 1:
|
||
return 1
|
||
return 0
|
||
|
||
# ============================================================
|
||
# 检查模块是否已导入或全局已登记（包子模块限定解析用）
|
||
# ============================================================
|
||
def is_module_imported_or_known(self, imported_modules: str, name: str) -> int:
|
||
"""已导入或可访问的模块（包子模块限定解析用）。
|
||
|
||
包子模块场景：`import intr` 只注册包名 "intr"，但 intr.gdt / intr.idt
|
||
等子模块由包 __init__.vp 的 from . import X 在全局注册——模块限定
|
||
调用/属性访问（intr.gdt.init()）的 is_module_imported 校验必须放行，
|
||
否则解析退化到函数名后缀匹配劫持其他模块同名函数
|
||
（serial.init 劫持 intr.gdt.init → GDT/IDT 未初始化根因）。
|
||
|
||
放行条件（保持单层名 fast-fail 语义不变）：
|
||
1. 名字已在 _imported_modules（直接 import）；
|
||
2. 名字在 Sha1Store 全局已登记，且其任一祖先包名已导入
|
||
（import intr → intr.gdt 可访问；`this.this()` 但 import this
|
||
被注释 → this 未导入 → 仍 fast-fail 报"模块未导入"）。
|
||
"""
|
||
if self.is_module_imported(imported_modules, name) != 0:
|
||
return 1
|
||
if name is None:
|
||
return 0
|
||
if StubMergerStore.LookupSha1ByModName(name) is None:
|
||
return 0
|
||
nl: t.CSizeT = string.strlen(name)
|
||
dot_pos: t.CSizeT = 0
|
||
while dot_pos < nl:
|
||
if name[dot_pos] == '.':
|
||
pfx: str = stdlib.malloc(dot_pos + 1)
|
||
if pfx is not None:
|
||
string.strncpy(pfx, name, dot_pos)
|
||
pfx[dot_pos] = '\0'
|
||
hit: int = self.is_module_imported(imported_modules, pfx)
|
||
stdlib.free(pfx)
|
||
if hit != 0:
|
||
return 1
|
||
dot_pos += 1
|
||
return 0
|
||
|
||
# ============================================================
|
||
# 添加 from-import 名称映射
|
||
# ============================================================
|
||
def add_from_import(self, pool: memhub.MemBuddy | t.CPtr,
|
||
from_imports: str,
|
||
local_name: str,
|
||
module_name: str,
|
||
original_name: str = None) -> str:
|
||
"""添加 from-import 映射，返回新的 from_imports 字符串
|
||
|
||
格式:
|
||
无别名: "local_name:module_name"
|
||
有别名: "local_name:module_name:original_name"
|
||
（original_name 为源模块中的真实函数名，用于跨模块 SHA1 修饰）
|
||
"""
|
||
if local_name is None or module_name is None:
|
||
return from_imports
|
||
entry: str = pool.alloc(256)
|
||
if entry is None:
|
||
return from_imports
|
||
if original_name is not None and string.strcmp(original_name, local_name) != 0:
|
||
viperlib.snprintf(entry, 256, "%s:%s:%s", local_name, module_name, original_name)
|
||
else:
|
||
viperlib.snprintf(entry, 256, "%s:%s", local_name, module_name)
|
||
if from_imports is None:
|
||
return entry
|
||
else:
|
||
old_len: t.CSizeT = string.strlen(from_imports)
|
||
entry_len: t.CSizeT = string.strlen(entry)
|
||
new_len: t.CSizeT = old_len + 1 + entry_len
|
||
buf: str = pool.alloc(new_len + 1)
|
||
if buf is not None:
|
||
string.strcpy(buf, from_imports)
|
||
buf[old_len] = ' '
|
||
string.strcpy(buf + old_len + 1, entry)
|
||
return buf
|
||
return from_imports
|
||
|
||
# ============================================================
|
||
# 查找 from-import 名称 → 返回原始函数名（别名场景）
|
||
#
|
||
# 对于 from X import Y as Z，返回 Y（源模块中的真实函数名）。
|
||
# 无别名时返回 None（local_name 即为原始名）。
|
||
# ============================================================
|
||
def lookup_from_import_original(self, from_imports: str,
|
||
local_name: str) -> str:
|
||
"""查找 from-import 别名对应的原始函数名，返回 None=无别名或未找到"""
|
||
if local_name is None or from_imports is None:
|
||
return None
|
||
name_len: t.CSizeT = string.strlen(local_name)
|
||
cur: str = from_imports
|
||
ci: t.CSizeT = 0
|
||
total_len: t.CSizeT = string.strlen(from_imports)
|
||
while ci < total_len:
|
||
# 跳过前导空格
|
||
while ci < total_len and cur[ci] == ' ':
|
||
ci += 1
|
||
if ci >= total_len:
|
||
break
|
||
# 找到第一个 ':' 的位置
|
||
colon1: t.CSizeT = ci
|
||
while colon1 < total_len and cur[colon1] != ':' and cur[colon1] != ' ':
|
||
colon1 += 1
|
||
if colon1 >= total_len or cur[colon1] != ':':
|
||
break
|
||
entry_name_len: t.CSizeT = colon1 - ci
|
||
# 跳过 star import
|
||
if entry_name_len == 1 and cur[ci] == '*':
|
||
ci = colon1
|
||
while ci < total_len and cur[ci] != ' ':
|
||
ci += 1
|
||
continue
|
||
# 比较名称
|
||
if entry_name_len == name_len:
|
||
match: int = 1
|
||
ei: t.CSizeT = 0
|
||
while ei < name_len:
|
||
if cur[ci + ei] != local_name[ei]:
|
||
match = 0
|
||
break
|
||
ei += 1
|
||
if match == 1:
|
||
# 找到匹配，检查是否有第三个字段（原始名）
|
||
pos: t.CSizeT = colon1 + 1
|
||
# 跳过模块名
|
||
while pos < total_len and cur[pos] != ':' and cur[pos] != ' ' and cur[pos] != '\0':
|
||
pos += 1
|
||
if pos < total_len and cur[pos] == ':':
|
||
# 有第三个字段: original_name
|
||
orig_start: t.CSizeT = pos + 1
|
||
orig_end: t.CSizeT = orig_start
|
||
while orig_end < total_len and cur[orig_end] != ' ' and cur[orig_end] != '\0':
|
||
orig_end += 1
|
||
orig_len: t.CSizeT = orig_end - orig_start
|
||
if orig_len > 0:
|
||
orig_buf: str = cur + orig_start
|
||
# 返回指向内部的指针（调用方需在使用期内保持 from_imports 有效）
|
||
return orig_buf
|
||
return None
|
||
return None
|
||
# 跳到下一个条目
|
||
ci = colon1
|
||
while ci < total_len and cur[ci] != ' ':
|
||
ci += 1
|
||
return None
|
||
|
||
# ============================================================
|
||
# 查找 from-import 名称 → 返回模块名或 None
|
||
#
|
||
# allow_star_fallback: 是否允许 star import 回退（默认 1=允许）。
|
||
# 当查询明确模块名（如 "BuildPipeline"）时，应传 0 禁用回退，
|
||
# 避免被 star import 模块名误导（如 from stdint import * 后
|
||
# 查询 "BuildPipeline" 错误回退到 "stdint"）。
|
||
# ============================================================
|
||
def lookup_from_import(self, from_imports: str, name: str,
|
||
allow_star_fallback: int = 1) -> str:
|
||
"""查找 from-import 名称，返回模块名或 None
|
||
|
||
支持 star import: 如果 from_imports 中有 "*:module" 条目，
|
||
且未找到精确名称匹配，且 allow_star_fallback != 0，则返回 star import 的模块名。
|
||
"""
|
||
if name is None or from_imports is None:
|
||
return None
|
||
name_len: t.CSizeT = string.strlen(name)
|
||
cur: str = from_imports
|
||
ci: t.CSizeT = 0
|
||
total_len: t.CSizeT = string.strlen(from_imports)
|
||
star_mod: str = None
|
||
while ci < total_len:
|
||
# 跳过前导空格
|
||
while ci < total_len and cur[ci] == ' ':
|
||
ci += 1
|
||
if ci >= total_len:
|
||
break
|
||
# 找到 ':' 的位置
|
||
colon_pos: t.CSizeT = ci
|
||
while colon_pos < total_len and cur[colon_pos] != ':' and cur[colon_pos] != ' ':
|
||
colon_pos += 1
|
||
if colon_pos >= total_len or cur[colon_pos] != ':':
|
||
break
|
||
entry_name_len: t.CSizeT = colon_pos - ci
|
||
# 检测 star import ("*:module")
|
||
if entry_name_len == 1 and cur[ci] == '*':
|
||
mod_start: t.CSizeT = colon_pos + 1
|
||
mod_end: t.CSizeT = mod_start
|
||
while mod_end < total_len and cur[mod_end] != ' ' and cur[mod_end] != '\0' and cur[mod_end] != ':':
|
||
mod_end += 1
|
||
star_mod = cur + mod_start
|
||
# 比较名称
|
||
elif entry_name_len == name_len:
|
||
match: int = 1
|
||
ei: t.CSizeT = 0
|
||
while ei < name_len:
|
||
if cur[ci + ei] != name[ei]:
|
||
match = 0
|
||
break
|
||
ei += 1
|
||
if match == 1:
|
||
mod_start2: t.CSizeT = colon_pos + 1
|
||
mod_end2: t.CSizeT = mod_start2
|
||
# 模块名结束于: 空格、null、或第二个':'（别名格式的分隔符）
|
||
while mod_end2 < total_len and cur[mod_end2] != ' ' and cur[mod_end2] != '\0' and cur[mod_end2] != ':':
|
||
mod_end2 += 1
|
||
return cur + mod_start2
|
||
# 跳到下一个条目
|
||
ci = colon_pos
|
||
while ci < total_len and cur[ci] != ' ':
|
||
ci += 1
|
||
# 未找到精确匹配，仅在允许时回退到 star import
|
||
if allow_star_fallback == 0:
|
||
return None
|
||
return star_mod
|
||
|
||
# ============================================================
|
||
# register_reexport - 注册 re-export 映射
|
||
#
|
||
# 在 HandleImportFromNames 中调用：当处理 from .X import Y 时，
|
||
# 记录当前模块（通过 ModuleSha1）re-export 了函数 Y，源模块为 X。
|
||
#
|
||
# 幂等：重复注册相同映射不会增加条目。
|
||
# 内存用 stdlib.malloc 分配（全局存储器，跨 Phase 持久化）。
|
||
# ============================================================
|
||
def register_reexport(self, mod_sha1: str, func_name: str, source_module: str) -> int:
|
||
"""注册 re-export 映射（幂等），返回 0 成功，-1 失败"""
|
||
if mod_sha1 is None or func_name is None or source_module is None:
|
||
return -1
|
||
# 懒初始化
|
||
if self.ReexportSha1s is None:
|
||
self.ReexportSha1s = CompilerFactory.get_perm_pool().alloc(MAX_REEXPORT * 17)
|
||
self.ReexportFuncs = CompilerFactory.get_perm_pool().alloc(MAX_REEXPORT * 64)
|
||
self.ReexportSrcs = CompilerFactory.get_perm_pool().alloc(MAX_REEXPORT * 64)
|
||
if self.ReexportSha1s is None or self.ReexportFuncs is None or self.ReexportSrcs is None:
|
||
return -1
|
||
string.memset(self.ReexportSha1s, 0, MAX_REEXPORT * 17)
|
||
string.memset(self.ReexportFuncs, 0, MAX_REEXPORT * 64)
|
||
string.memset(self.ReexportSrcs, 0, MAX_REEXPORT * 64)
|
||
if self.ReexportCount >= MAX_REEXPORT:
|
||
return -1
|
||
# 幂等检查：查找是否已存在相同映射
|
||
for i in range(self.ReexportCount):
|
||
sidx: t.CSizeT = t.CSizeT(i) * 17
|
||
fidx: t.CSizeT = t.CSizeT(i) * 64
|
||
if string.strcmp(self.ReexportSha1s + sidx, mod_sha1) == 0:
|
||
if string.strcmp(self.ReexportFuncs + fidx, func_name) == 0:
|
||
# 已存在，更新 source_module（以防变化；64 字节槽，strncpy 钳制）
|
||
string.strncpy(self.ReexportSrcs + fidx, source_module, 63)
|
||
self.ReexportSrcs[fidx + 63] = '\0'
|
||
return 0
|
||
# 添加新条目（槽位定长 17/64 字节，strcpy 无界会越界写相邻槽，全部 strncpy 钳制）
|
||
idx2: t.CSizeT = t.CSizeT(self.ReexportCount) * 17
|
||
fidx2: t.CSizeT = t.CSizeT(self.ReexportCount) * 64
|
||
string.strncpy(self.ReexportSha1s + idx2, mod_sha1, 16)
|
||
self.ReexportSha1s[idx2 + 16] = '\0'
|
||
string.strncpy(self.ReexportFuncs + fidx2, func_name, 63)
|
||
self.ReexportFuncs[fidx2 + 63] = '\0'
|
||
string.strncpy(self.ReexportSrcs + fidx2, source_module, 63)
|
||
self.ReexportSrcs[fidx2 + 63] = '\0'
|
||
self.ReexportCount += 1
|
||
return 0
|
||
|
||
# ============================================================
|
||
# lookup_reexport - 查找 re-export 映射
|
||
#
|
||
# 给定模块 SHA1 和函数名，返回源模块名（如 "ast.lexer"）或 None。
|
||
# 调用方通过 _lookup_module_sha1(source_module) 获取源模块的 SHA1。
|
||
# ============================================================
|
||
def lookup_reexport(self, mod_sha1: str, func_name: str) -> str:
|
||
"""查找 re-export 映射，返回源模块名或 None"""
|
||
if mod_sha1 is None or func_name is None:
|
||
return None
|
||
if self.ReexportSha1s is None or self.ReexportCount <= 0:
|
||
return None
|
||
for i in range(self.ReexportCount):
|
||
sidx: t.CSizeT = t.CSizeT(i) * 17
|
||
fidx: t.CSizeT = t.CSizeT(i) * 64
|
||
if string.strcmp(self.ReexportSha1s + sidx, mod_sha1) == 0:
|
||
if string.strcmp(self.ReexportFuncs + fidx, func_name) == 0:
|
||
return self.ReexportSrcs + fidx
|
||
return None
|
||
|
||
# ============================================================
|
||
# _resolve_relative_module - 解析相对导入为完整模块名
|
||
#
|
||
# 对于 from .X import Y（level=1, module="X"）在包 pkg 中：
|
||
# 解析为 "pkg.X"
|
||
# 对于 from . import Y（level=1, module=None）在包 pkg 中：
|
||
# 解析为 "pkg"
|
||
# 对于 from ..X import Y（level=2, module="X"）在包 pkg.sub 中：
|
||
# 解析为 "pkg.X"（先从 pkg.sub 上溯一级到 pkg，再追加 .X）
|
||
#
|
||
# Args:
|
||
# pool: 内存池
|
||
# current_package: 当前文件所属包名（如 "llvmlite"），None 表示无包
|
||
# level: 相对导入级别（0=绝对，1=., 2=..）
|
||
# module: ImportFrom 的 module 字段（可能为 None）
|
||
#
|
||
# Returns:
|
||
# 解析后的完整模块名；绝对导入（level<=0）直接返回 module；
|
||
# 无法解析时返回 module（回退到原始值）
|
||
# ============================================================
|
||
def _resolve_relative_module(self, pool: memhub.MemBuddy | t.CPtr,
|
||
current_package: str,
|
||
level: t.CInt,
|
||
module: str) -> str:
|
||
"""解析相对导入为完整模块名"""
|
||
if level <= 0:
|
||
return module
|
||
if current_package is None:
|
||
# 无包上下文（includes 顶级文件 / App 顶层文件）：
|
||
# `from .X import Y` 的语义 = "同目录/根目录的 X 模块"
|
||
# （如 includes/viperstring.vp 的 `from .string import *`）。
|
||
# 直接返回 module 是正确解析，不是降级——禁止 fail-fast。
|
||
return module
|
||
|
||
# 从 current_package 开始，上溯 (level-1) 级
|
||
pkg: str = current_package
|
||
up: t.CInt = level - 1
|
||
while up > 0:
|
||
pkg_len: t.CSizeT = string.strlen(pkg)
|
||
last_dot: t.CSizeT = 0
|
||
found: int = 0
|
||
i: t.CSizeT = 0
|
||
while i < pkg_len:
|
||
if pkg[i] == '.':
|
||
last_dot = i
|
||
found = 1
|
||
i += 1
|
||
if found == 0:
|
||
# 无更多父级（相对导入级别超过包深度，上溯到根目录）：
|
||
# 回退到 module 原始值（根目录模块语义），后续查找失败
|
||
# 由 import SHA1 检查明确报"模块未找到"，此处不误杀。
|
||
pkg = None
|
||
break
|
||
# 截断到最后一个 '.' 处
|
||
new_pkg: str = pool.alloc(last_dot + 1)
|
||
if new_pkg is None:
|
||
self.Gen.TypeGenErr.fatal_error(None, f"相对导入解析失败：level={level} module='{module}'", "ImportError")
|
||
return module
|
||
string.strncpy(new_pkg, pkg, last_dot)
|
||
new_pkg[last_dot] = '\0'
|
||
pkg = new_pkg
|
||
up -= 1
|
||
|
||
if module is None:
|
||
# from . import Y → 模块就是包本身
|
||
return pkg
|
||
if pkg is None:
|
||
# 包已上溯到空（相对导入越界）：回退到 module 原始值
|
||
return module
|
||
# 拼接 pkg + "." + module
|
||
pkg_len2: t.CSizeT = string.strlen(pkg)
|
||
mod_len: t.CSizeT = string.strlen(module)
|
||
buf: str = pool.alloc(pkg_len2 + 1 + mod_len + 1)
|
||
if buf is None:
|
||
self.Gen.TypeGenErr.fatal_error(None, f"相对导入解析失败：level={level} module='{module}'", "ImportError")
|
||
return module
|
||
string.strcpy(buf, pkg)
|
||
buf[pkg_len2] = '.'
|
||
string.strcpy(buf + pkg_len2 + 1, module)
|
||
return buf
|
||
|
||
# ============================================================
|
||
# compute_package_from_relpath - 从相对路径计算包名
|
||
#
|
||
# 包名 = 文件所在目录路径，将 / 和 \ 替换为 .
|
||
# 对于顶级文件（无目录分隔符），返回 None
|
||
#
|
||
# 示例：
|
||
# "llvmlite/__init__.vp" → "llvmlite"
|
||
# "llvmlite/__types.vp" → "llvmlite"
|
||
# "ast/parser.vp" → "ast"
|
||
# "ast.vp" → None（顶级文件）
|
||
# ============================================================
|
||
def compute_package_from_relpath(self, pool: memhub.MemBuddy | t.CPtr,
|
||
rel_path: str) -> str:
|
||
"""从相对路径计算包名（目录部分，分隔符替换为 .）"""
|
||
if rel_path is None:
|
||
return None
|
||
rlen: t.CSizeT = string.strlen(rel_path)
|
||
# 找最后一个 / 或 \
|
||
last_sep: t.CSizeT = 0
|
||
found: int = 0
|
||
i: t.CSizeT = 0
|
||
while i < rlen:
|
||
ch: t.CChar = rel_path[i]
|
||
if ch == '/' or ch == '\\':
|
||
last_sep = i
|
||
found = 1
|
||
i += 1
|
||
if found == 0:
|
||
# 无目录分隔符 → 顶级文件，无包
|
||
return None
|
||
# 复制目录部分，将 / 和 \ 替换为 .
|
||
dir_len: t.CSizeT = last_sep
|
||
buf: str = pool.alloc(dir_len + 1)
|
||
if buf is None:
|
||
return None
|
||
j: t.CSizeT = 0
|
||
while j < dir_len:
|
||
ch2: t.CChar = rel_path[j]
|
||
if ch2 == '/' or ch2 == '\\':
|
||
buf[j] = '.'
|
||
else:
|
||
buf[j] = ch2
|
||
j += 1
|
||
buf[dir_len] = '\0'
|
||
return buf
|
||
|
||
# ============================================================
|
||
# set_current_file - 设置当前编译的文件名
|
||
#
|
||
# 与 TypeGenMixin.set_current_file 行为一致（写 CurrentFile 字段），
|
||
# BuildPipeline/Phase1 通过 HandlesImports.set_current_file 委托调用。
|
||
# 经 HandlesType 委托（延迟 import 规避循环依赖），字段由 TypeGenMixin 持有。
|
||
# ============================================================
|
||
def set_current_file(self, file_path: str) -> None:
|
||
"""设置当前正在编译的文件名（与 HandlesType.set_current_file 同步调用）"""
|
||
self.Gen.TypeGenResolve.set_current_file(file_path)
|
||
|
||
# ============================================================
|
||
# _report_import_error - 本地实现 Python 风格报错（避免循环依赖）
|
||
#
|
||
# 无法访问 HandlesType.fatal_error（HandlesType import 了本模块，
|
||
# 若本模块再 import HandlesType 会形成循环依赖），这里用简化格式：
|
||
# Traceback (most recent call last):
|
||
# File "<当前文件>", line <N>
|
||
# ImportError: <msg>
|
||
# ============================================================
|
||
def _report_import_error(self, node: ast.AST | t.CPtr, msg: str) -> None:
|
||
"""以 Python 风格报错并退出（本地实现，避免循环依赖）"""
|
||
stdio.printf("Traceback (most recent call last):\n")
|
||
lineno: int = 0
|
||
if node is not None:
|
||
lineno = node.lineno
|
||
file_name: str = self.Gen.TypeGenResolve.get_current_file()
|
||
if file_name is None:
|
||
file_name = "<unknown>"
|
||
stdio.printf(' File "%s", line %d\n', file_name, lineno)
|
||
# 显示行内内容（读源文件第 lineno 行的实际代码，便于快速定位）
|
||
if lineno > 0 and file_name is not None:
|
||
f_err: fileio.File | t.CPtr = fileio.File(file_name, fileio.MODE.R)
|
||
if not f_err.closed:
|
||
line_buf_err: bytes = stdlib.malloc(512)
|
||
if line_buf_err is not None:
|
||
cur_ln: int = 0
|
||
while cur_ln < lineno:
|
||
r_err: LONG = f_err.readline(line_buf_err, 511)
|
||
if r_err <= 0:
|
||
break
|
||
cur_ln += 1
|
||
if cur_ln == lineno:
|
||
ll_err: t.CSizeT = string.strlen(line_buf_err)
|
||
while ll_err > 0 and (line_buf_err[ll_err - 1] == '\n' or line_buf_err[ll_err - 1] == '\r'):
|
||
ll_err -= 1
|
||
line_buf_err[ll_err] = '\0'
|
||
stdio.printf(" %s\n", line_buf_err)
|
||
break
|
||
stdlib.free(line_buf_err)
|
||
f_err.close()
|
||
stdio.printf("ImportError: %s\n", msg)
|
||
sys.exit(1)
|
||
|
||
# ============================================================
|
||
# _is_local_module_file - 当前文件同目录存在 {mod_name}.vp/.py
|
||
#
|
||
# 多文件项目（如 Test/G4/App 的 caller1.vp import g4node）的兄弟模块
|
||
# 不在 includes 扫描存储器中（SHA1 在项目源文件列表），检查应豁免。
|
||
# ============================================================
|
||
def _is_local_module_file(self, mod_name: str) -> int:
|
||
"""当前文件同目录存在 {mod_name}.vp/.py → 项目内兄弟模块"""
|
||
cur: str = self.Gen.TypeGenResolve.get_current_file()
|
||
if cur is None or mod_name is None:
|
||
return 0
|
||
cur_len: t.CSizeT = string.strlen(cur)
|
||
if cur_len == 0:
|
||
return 0
|
||
sep_pos: t.CSizeT = cur_len
|
||
while sep_pos > 0:
|
||
sep_pos -= 1
|
||
ch: t.CChar = cur[sep_pos]
|
||
if ch == '/' or ch == '\\':
|
||
sep_pos += 1
|
||
break
|
||
mod_len: t.CSizeT = string.strlen(mod_name)
|
||
path_len: t.CSizeT = sep_pos + mod_len + 5
|
||
path_buf: bytes = stdlib.malloc(path_len + 1)
|
||
if path_buf is None:
|
||
return 0
|
||
# .vp 文件（Viper 源仅 .vp，不再回退 .py）
|
||
string.strncpy(path_buf, cur, sep_pos)
|
||
path_buf[sep_pos] = '\0'
|
||
string.strcat(path_buf, mod_name)
|
||
string.strcat(path_buf, ".vp")
|
||
fex: fileio.File | t.CPtr = fileio.File(path_buf, fileio.MODE.R)
|
||
if not fex.closed:
|
||
fex.close()
|
||
stdlib.free(path_buf)
|
||
return 1
|
||
stdlib.free(path_buf)
|
||
return 0
|
||
|
||
# ============================================================
|
||
# _check_import_sha1 - 导入时校验模块 SHA1 是否已注册（fast-fail）
|
||
#
|
||
# 背景：若 includes 路径配置错误（如 project.json 指向不存在的
|
||
# ./includes），includes 扫描结果为 0，模块不会注册 SHA1。此前
|
||
# 错误延迟到实际调用 module.func() 时才报误导性的
|
||
# "跨模块调用 'bd.alloc' 找不到模块 SHA1"。
|
||
# 现在在 import 语句处立即检测并报错，快速定位配置问题。
|
||
#
|
||
# t/c 是内部库，不参与 includes 扫描，豁免检查。
|
||
# ============================================================
|
||
def _check_import_sha1(self, node: ast.AST | t.CPtr, mod_name: str) -> int:
|
||
"""导入时校验模块 SHA1 是否已注册，未注册则立即报错退出
|
||
|
||
Args:
|
||
node: Import/ImportFrom AST 节点（提供行号）
|
||
mod_name: 导入的模块名（如 "memhub"、"w32.fileio"）
|
||
|
||
Returns:
|
||
1=检查通过；模块未注册时直接报错退出（不会返回 0）
|
||
"""
|
||
if mod_name is None:
|
||
return 1
|
||
# t/c 是内部库，不参与 includes 扫描，豁免检查
|
||
# ctraits 是纯编译期宏库（.vpi 后缀，IncludesScanner 不扫描 .vpi），豁免检查
|
||
if string.strcmp(mod_name, "t") == 0 or string.strcmp(mod_name, "c") == 0 or string.strcmp(mod_name, "ctraits") == 0:
|
||
return 1
|
||
mod_sha1: str = StubMergerStore.LookupSha1ByModName(mod_name)
|
||
if mod_sha1 is None:
|
||
# App 兄弟模块豁免：当前文件同目录存在 {mod_name}.vp/.py
|
||
# （如 Test/G4/App/caller1.vp import g4node——同项目源文件，
|
||
# SHA1 在项目文件列表而非 includes 扫描存储器，不应报错）
|
||
if self._is_local_module_file(mod_name) != 0:
|
||
return 1
|
||
err_buf: str = stdlib.malloc(256)
|
||
if err_buf is not None:
|
||
viperlib.snprintf(err_buf, 256,
|
||
"模块 '%s' 未注册 SHA1（includes 扫描未找到该模块，请检查 project.json 的 includes 路径配置）",
|
||
mod_name)
|
||
self._report_import_error(node, err_buf)
|
||
else:
|
||
self._report_import_error(node, "导入模块 SHA1 未注册（includes 扫描未找到该模块）")
|
||
return 1
|
||
|
||
# ============================================================
|
||
# translate_import - 处理 import 语句，更新 trans._imported_modules
|
||
# （迁移自 HandlesImports.ImportsHandle.HandleImport）
|
||
# ============================================================
|
||
def translate_import(self, trans: HT.Translator | t.CPtr,
|
||
node: ast.AST | t.CPtr) -> int:
|
||
"""处理 import 语句，返回 0"""
|
||
imp: ast.Import | t.CPtr = (ast.Import | t.CPtr)(node)
|
||
if imp is None:
|
||
return 0
|
||
names: list[ast.AST | t.CPtr] | t.CPtr = imp.names
|
||
if names is None:
|
||
return 0
|
||
nn: t.CSizeT = len(names)
|
||
for ni in range(nn):
|
||
alias: ast.Alias | t.CPtr = (ast.Alias | t.CPtr)(names.get(ni))
|
||
if alias is not None and alias.name is not None:
|
||
# 导入时立即校验模块 SHA1 是否注册（fast-fail，仅全量翻译时检查）
|
||
# declare_only=1/2 是 struct 预注册/import 扫描阶段：此时 SHA1 表
|
||
# 尚未包含后续才追加的模块（如 App 源文件），检查会造成误报
|
||
if trans._declare_only == 0:
|
||
self._check_import_sha1(node, alias.name)
|
||
pool_val: memhub.MemBuddy | t.CPtr = trans.Pool
|
||
im_val: str = trans._imported_modules
|
||
trans._imported_modules = self.add_imported_module(
|
||
pool_val, im_val, alias.name)
|
||
# 别名也加入导入模块列表（用于模块限定构造器检查 Module.Class()）
|
||
if alias.asname is not None:
|
||
trans._imported_modules = self.add_imported_module(
|
||
pool_val, trans._imported_modules, alias.asname)
|
||
# 别名也加入 from_imports（asname:fullname），
|
||
# 使 HT.func() 跨模块调用能通过 from_imports 查找别名→完整模块名→SHA1
|
||
# 否则 _lookup_module_sha1("HT") 找不到（mod_arr 只存文件名 HandlesTranslator）
|
||
trans._from_imports = self.add_from_import(
|
||
pool_val, trans._from_imports, alias.asname, alias.name)
|
||
else:
|
||
# 无别名时（import a.b.c），将顶层包名 "a" 也加入 _imported_modules，
|
||
# 使 a.b.c 作为属性访问的 base 被翻译时能通过 is_module_imported("a") 检查
|
||
dot_ptr: str = string.strstr(alias.name, ".")
|
||
if dot_ptr is not None:
|
||
pkg_len: t.CSizeT = t.CSizeT(t.CUInt64T(dot_ptr) - t.CUInt64T(alias.name))
|
||
if pkg_len > 0 and pkg_len < 256:
|
||
pkg_buf: bytes = pool_val.alloc(pkg_len + 1)
|
||
if pkg_buf is not None:
|
||
string.strncpy(pkg_buf, alias.name, pkg_len)
|
||
pkg_buf[pkg_len] = '\0'
|
||
trans._imported_modules = self.add_imported_module(
|
||
pool_val, trans._imported_modules, pkg_buf)
|
||
return 0
|
||
|
||
# ============================================================
|
||
# translate_import_from_module - 处理 from X import Y 的模块部分
|
||
# （迁移自 HandlesImports.ImportsHandle.HandleImportFromModule）
|
||
#
|
||
# 对相对导入（level > 0），使用 trans.CurrentPackage 解析为
|
||
# 完整模块名（如 __types → llvmlite.__types），确保 .deps.txt
|
||
# 记录的模块名与 SHA1 映射表一致。
|
||
# ============================================================
|
||
def translate_import_from_module(self, trans: HT.Translator | t.CPtr,
|
||
node: ast.AST | t.CPtr) -> int:
|
||
"""处理 from X import Y 的模块部分，更新 trans._imported_modules"""
|
||
impf: ast.ImportFrom | t.CPtr = (ast.ImportFrom | t.CPtr)(node)
|
||
if impf is None:
|
||
return 0
|
||
# 解析模块名：相对导入需补充包前缀
|
||
resolved: str = self._resolve_relative_module(
|
||
trans.Pool, trans.CurrentPackage,
|
||
impf.level, impf.module)
|
||
if resolved is not None:
|
||
# 导入时立即校验模块 SHA1 是否注册（fast-fail，仅全量翻译时检查）
|
||
if trans._declare_only == 0:
|
||
self._check_import_sha1(node, resolved)
|
||
trans._imported_modules = self.add_imported_module(
|
||
trans.Pool, trans._imported_modules, resolved)
|
||
# from . import Y（impf.module is None）：resolved 是包名（vpsdk），
|
||
# 各 name 是包的子模块 vpsdk.Y。须把完整子模块名也加入
|
||
# _imported_modules，否则别名 "syscall" 精确匹配不到包名 "vpsdk"
|
||
# → is_module_imported 返回 0 → syscall._syscall1 / syscall.LOAD_SO
|
||
# 跨模块解析失败（vpsdk/dynlib.vp 根因）。
|
||
if impf.module is None:
|
||
names_m: list[ast.AST | t.CPtr] | t.CPtr = impf.names
|
||
if names_m is not None:
|
||
nn_m: t.CSizeT = len(names_m)
|
||
for ni_m in range(nn_m):
|
||
alias_m: ast.Alias | t.CPtr = (ast.Alias | t.CPtr)(names_m.get(ni_m))
|
||
if alias_m is not None and alias_m.name is not None:
|
||
rl_m: t.CSizeT = string.strlen(resolved)
|
||
rn_m: t.CSizeT = string.strlen(alias_m.name)
|
||
jb_m: str = trans.Pool.alloc(rl_m + rn_m + 2)
|
||
if jb_m is not None:
|
||
string.strcpy(jb_m, resolved)
|
||
jb_m[rl_m] = '.'
|
||
string.strcpy(jb_m + rl_m + 1, alias_m.name)
|
||
if trans._declare_only == 0:
|
||
self._check_import_sha1(node, jb_m)
|
||
trans._imported_modules = self.add_imported_module(
|
||
trans.Pool, trans._imported_modules, jb_m)
|
||
return 0
|
||
|
||
# ============================================================
|
||
# translate_import_from_names - 处理 from X import Y 的名称部分
|
||
# （迁移自 HandlesImports.ImportsHandle.HandleImportFromNames）
|
||
#
|
||
# 同样使用解析后的完整模块名，确保 from-import 映射
|
||
# （如 LLVMType:llvmlite.__types）能正确查到 SHA1。
|
||
# ============================================================
|
||
def translate_import_from_names(self, trans: HT.Translator | t.CPtr,
|
||
node: ast.AST | t.CPtr) -> int:
|
||
"""处理 from X import Y 的名称部分，更新 trans._from_imports"""
|
||
impf: ast.ImportFrom | t.CPtr = (ast.ImportFrom | t.CPtr)(node)
|
||
if impf is None:
|
||
return 0
|
||
# 解析模块名：相对导入需补充包前缀
|
||
resolved: str = self._resolve_relative_module(
|
||
trans.Pool, trans.CurrentPackage,
|
||
impf.level, impf.module)
|
||
if resolved is not None:
|
||
names: list[ast.AST | t.CPtr] | t.CPtr = impf.names
|
||
if names is not None:
|
||
nn: t.CSizeT = len(names)
|
||
for ni in range(nn):
|
||
alias: ast.Alias | t.CPtr = (ast.Alias | t.CPtr)(names.get(ni))
|
||
if alias is not None and alias.name is not None:
|
||
local_name: str = alias.name
|
||
orig_name: str = None
|
||
if alias.asname is not None:
|
||
local_name = alias.asname
|
||
orig_name = alias.name
|
||
# from . import Y（impf.module is None）：_resolve_relative_module
|
||
# 返回的是包名（如 vpsdk），但 Y 是包的子模块 vpsdk.Y
|
||
# （vpsdk/syscall.vp）。此前直接映射 local_name → 包名，
|
||
# 导致 syscall.LOAD_SO 跨模块 CDefine / syscall._syscall1
|
||
# 跨模块调用在包里找不到（vpsdk/dynlib.vp `from . import
|
||
# syscall as syscall` 根因）。
|
||
fi_target: str = resolved
|
||
if impf.module is None and resolved is not None:
|
||
fl_t: t.CSizeT = string.strlen(resolved)
|
||
fn_t: t.CSizeT = string.strlen(alias.name)
|
||
jb_t: str = trans.Pool.alloc(fl_t + fn_t + 2)
|
||
if jb_t is not None:
|
||
string.strcpy(jb_t, resolved)
|
||
jb_t[fl_t] = '.'
|
||
string.strcpy(jb_t + fl_t + 1, alias.name)
|
||
fi_target = jb_t
|
||
trans._from_imports = self.add_from_import(
|
||
trans.Pool, trans._from_imports,
|
||
local_name, fi_target, orig_name)
|
||
# 命名空间隔离：from-import 的名称标记为可见结构体
|
||
self.Gen.StructGen.add_visible_struct(trans.Pool, local_name)
|
||
# 注册 re-export 映射：当前模块 re-export 了 local_name，
|
||
# 源模块为 fi_target（from . import Y 时是子模块名 vpsdk.Y，
|
||
# 非包名）。使跨模块调用 module.func() 能解析到正确 SHA1
|
||
if trans.ModuleSha1 is not None:
|
||
self.register_reexport(trans.ModuleSha1, local_name, fi_target)
|
||
return 0
|
||
|
||
def InitImportsGen(self) -> int:
|
||
"""初始化 ImportsGen 字段（构造时由 Generator 调用）"""
|
||
if self is None:
|
||
return 1
|
||
self.ReexportSha1s = None
|
||
self.ReexportFuncs = None
|
||
self.ReexportSrcs = None
|
||
self.ReexportCount = 0
|
||
return 0
|