Files
TransPyC/TransPyV/App/lib/core/StubMerger.py

1530 lines
66 KiB
Python
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 memhub
import string
import stdio
import stdlib
import llvmlite
import w32.win32file as win32file
import w32.win32base as win32base
import w32.fileio as fileio
import w32.win32memory as win32memory
import viperlib
import sys
import ast
import lib.core.IncludesScanner as IncludesScanner
import lib.core.Handles.HandlesStruct as HandlesStruct
import lib.core.Handles.HandlesImports as HandlesImports
import lib.core.Handles.HandlesTranslator as HandlesTranslator
# ============================================================
# StubMerger - stub 声明合并模块
#
# 从 temp 目录加载所有 .stub.ll 文件,将声明内容合并到主 IR。
# 这是 Phase2 的核心功能:翻译主文件时,将所有依赖的 stub
# 声明合并到主 IR 前面,使链接器能解析跨模块符号引用。
#
# 合并策略:
# 1. 扫描 temp 目录,收集所有 {sha1}.stub.ll 文件
# 2. 读取每个 stub 文件内容
# 3. 过滤去重(避免重复声明)
# 4. 将声明插入到主 IR 的 target triple 之后
# ============================================================
# 全局 mbuddy 指针
_mbuddy: memhub.MemManager | t.CPtr
# 最大 includes SHA1 数(用于过滤)
MAX_INCLUDES_SHA1: t.CDefine = 256
# stub 读取缓冲区大小
STUB_READ_BUF_SIZE: t.CDefine = 262144
# 最大 includes 条目数
MAX_INCLUDES: t.CDefine = 256
# ============================================================
# _load_includes_sha1_set - 读取 _sha1_map.txt收集 includes/ 开头的 SHA1
#
# _sha1_map.txt 格式:{sha16}:{rel_path}
# 仅收集 rel_path 以 "includes/" 开头的 SHA1用于过滤 stub 文件,
# 避免加载 TransPyV 自身源文件lib/*)的 stub 导致链接失败。
# ============================================================
def _load_includes_sha1_set(pool: memhub.MemBuddy | t.CPtr,
temp_dir: str,
sha1_set: str) -> int:
"""读取 _sha1_map.txt将 includes/ 开头的 SHA1 写入 sha1_set
sha1_set 大小为 MAX_INCLUDES_SHA1 * 17每个 SHA1 16字符+null
返回找到的 includes SHA1 数量,-1 表示错误"""
if pool is None or temp_dir is None or sha1_set is None:
return -1
# 构造路径 temp_dir/_sha1_map.txt使用 stdlib.malloc 避免 mbuddy 池耗尽)
dir_len: t.CSizeT = string.strlen(temp_dir)
map_path: bytes = stdlib.malloc(dir_len + 32)
if map_path is None:
return -1
viperlib.snprintf(map_path, dir_len + 32, "%s/_sha1_map.txt", temp_dir)
# 打开文件
f: fileio.File | t.CPtr = fileio.File(map_path, fileio.MODE.R)
if f.closed:
stdio.printf("[StubMerger] _sha1_map.txt 不存在: %s\n", map_path)
stdlib.free(map_path)
return -1
# 读取内容(使用 stdlib.malloc 避免 mbuddy 池耗尽)
MAP_BUF_SIZE: t.CSizeT = 65536
content: str = stdlib.malloc(MAP_BUF_SIZE)
if content is None:
f.close()
stdlib.free(map_path)
return -1
bytes_read: t.CInt64T = f.read_all(content, MAP_BUF_SIZE)
f.close()
stdlib.free(map_path)
if bytes_read <= 0:
stdlib.free(content)
return -1
content[bytes_read] = '\0'
# 解析行: {sha1}:{rel_path}
count: int = 0
pos: t.CSizeT = 0
content_len: t.CSizeT = bytes_read
while pos < content_len:
# 找行尾
line_start: t.CSizeT = pos
while pos < content_len:
if content[pos] == '\n':
break
pos += 1
line_len: t.CSizeT = pos - line_start
pos += 1 # skip \n
# 最小长度: 16(sha1) + 1(:) + 9(includes/) = 26
if line_len < 26:
continue
# 检查第17个字符是 ':'
if content[line_start + 16] != ':':
continue
# 检查 rel_path 是否以 "includes/" 开头
rel_start: t.CSizeT = line_start + 17
if string.strncmp(content + rel_start, "includes/", 9) == 0:
if count < MAX_INCLUDES_SHA1:
string.strncpy(sha1_set + count * 17, content + line_start, 16)
sha1_set[count * 17 + 16] = '\0'
count += 1
stdlib.free(content)
return count
# ============================================================
# _is_in_sha1_set - 检查 SHA1 是否在集合中
# ============================================================
def _is_in_sha1_set(sha1: str, sha1_set: str, set_count: int) -> int:
"""检查 sha1 是否在集合中"""
if sha1 is None or sha1_set is None:
return 0
for i in range(set_count):
if string.strcmp(sha1_set + i * 17, sha1) == 0:
return 1
return 0
# ============================================================
# WriteIncludesSha1Map - 将 includes 扫描结果写入 _sha1_map.txt
#
# TransPyV.exe 自己生成 _sha1_map.txt不再依赖 Projectrans.py 预生成),
# 确保所有 includes 文件的 SHA1 都被记录,供 StubMerger 加载 stub 时过滤。
#
# Args:
# mb: 内存池
# temp_dir: 临时目录_sha1_map.txt 写入位置)
# scan_result: IncludesScanner 扫描结果
#
# Returns:
# 0 成功,非 0 失败
# ============================================================
def WriteIncludesSha1Map(mb: memhub.MemBuddy | t.CPtr, temp_dir: str,
scan_result: IncludesScanner.ScanResult | t.CPtr,
filter_set: t.CChar | t.CPtr,
filter_count: int) -> int:
"""将 includes 扫描结果写入 _sha1_map.txt供 StubMerger 使用)
若 filter_set 不为 None 且 filter_count > 0只写入 filter_set 中的条目。
"""
if temp_dir is None or scan_result is None:
return 1
# 构造路径 temp_dir/_sha1_map.txt
dir_len: t.CSizeT = string.strlen(temp_dir)
map_path: bytes = stdlib.malloc(dir_len + 32)
if map_path is None:
return 1
viperlib.snprintf(map_path, dir_len + 32, "%s/_sha1_map.txt", temp_dir)
# 打开文件写入CREATE_ALWAYS
f: fileio.File | t.CPtr = fileio.File(map_path, fileio.MODE.W)
if f.closed:
stdio.printf("[Phase1] 无法写入 _sha1_map.txt: %s\n", map_path)
return 1
# 写入每个 include 条目: {sha1}:includes/{rel_path}\n
entry_size_w: t.CSizeT = IncludesScanner.FileEntry.__sizeof__()
line_buf: bytes = stdlib.malloc(512)
if line_buf is None:
f.close()
return 1
written_count: int = 0
for i in range(scan_result.Count):
ea_w: t.CUInt64T = t.CUInt64T(scan_result.Entries) + i * entry_size_w
ent_w: IncludesScanner.FileEntry | t.CPtr = (IncludesScanner.FileEntry | t.CPtr)(t.CVoid(ea_w, t.CPtr))
if ent_w is None or ent_w.Sha1 is None or ent_w.RelPath is None:
continue
# 若提供过滤集合,只写入集合中的条目
if filter_set is not None and filter_count > 0:
if _is_in_sha1_set(ent_w.Sha1, filter_set, filter_count) == 0:
continue
viperlib.snprintf(line_buf, 512, "%s:includes/%s\n", ent_w.Sha1, ent_w.RelPath)
line_len_w: t.CSizeT = string.strlen(line_buf)
f.write(line_buf, line_len_w)
written_count += 1
f.close()
stdlib.free(map_path)
stdlib.free(line_buf)
stdio.printf("[Phase1] 已写入 _sha1_map.txt (%d 个 includes)\n", written_count)
return 0
# ============================================================
# _IsFuncDeclaredOrDefined - 检查 out_buf 中是否已有函数的 declare/define
#
# 逐字节strncmp搜索 func_sig如 @"sha1.func"(/ @printf( ),每次
# 找到匹配后检查行首是否以 "declare " 或 "define " 开头,跳过 call 语句。
# 避免 strstr 子串匹配误匹配 call ... @"func"(...) 中的 @"func"( 子串。
#
# Args:
# out_buf: 已有的 IR 文本缓冲区
# func_sig: 函数签名片段,如 @"sha1.func"(/ @printf(
#
# Returns:
# 1=已存在 declare/define, 0=不存在
# ============================================================
def _IsFuncDeclaredOrDefined(out_buf: bytes, func_sig: str) -> int:
"""检查 out_buf 中是否已有函数的 declare 或 define 行"""
if out_buf is None or func_sig is None:
return 0
sig_len: t.CSizeT = string.strlen(func_sig)
out_len: t.CSizeT = string.strlen(out_buf)
if sig_len == 0 or out_len == 0 or sig_len > out_len:
return 0
pos: t.CSizeT = 0
while pos + sig_len <= out_len:
if string.strncmp(out_buf + pos, func_sig, sig_len) == 0:
ls: t.CSizeT = pos
while ls > 0 and out_buf[ls - 1] != '\n':
ls -= 1
if string.strncmp(out_buf + ls, "declare ", 8) == 0:
return 1
if string.strncmp(out_buf + ls, "define ", 7) == 0:
return 1
pos += 1
return 0
# ============================================================
# _CommentOutDeclareInBuf - 将 out_buf 中匹配 func_sig 的 declare 行注释掉
#
# 用于解决 declare/define 冲突:当本地 text.ll 包含 define 而依赖
# text.ll 已生成了同函数的 declare 时,将 declare 行首字符改为 ';'
# 使其成为注释,然后追加 define 行。
# ============================================================
def _CommentOutDeclareInBuf(out_buf: bytes, func_sig: str) -> int:
"""在 out_buf 中搜索 declare 行(包含 func_sig将其注释掉"""
if out_buf is None or func_sig is None:
return 0
sig_len: t.CSizeT = string.strlen(func_sig)
out_len: t.CSizeT = string.strlen(out_buf)
if sig_len == 0 or out_len == 0 or sig_len > out_len:
return 0
pos: t.CSizeT = 0
while pos + sig_len <= out_len:
if string.strncmp(out_buf + pos, func_sig, sig_len) == 0:
ls: t.CSizeT = pos
while ls > 0 and out_buf[ls - 1] != '\n':
ls -= 1
if string.strncmp(out_buf + ls, "declare ", 8) == 0:
out_buf[ls] = ';'
return 1
pos += 1
return 0
# ============================================================
# _CommentOutGlobalInBuf - 将 out_buf 中匹配全局变量名的行注释掉
#
# 解决 stub/text 全局变量重复定义问题stub 中可能是 external 声明
# 或完整定义,以 text 中的定义为准,将 stub 中的同名行注释掉。
# ============================================================
def _CommentOutGlobalInBuf(out_buf: bytes, global_name: str) -> int:
"""在 out_buf 中搜索全局变量定义行(以 global_name 开头),将行首改为 ';'"""
if out_buf is None or global_name is None:
return 0
name_len: t.CSizeT = string.strlen(global_name)
out_len: t.CSizeT = string.strlen(out_buf)
if name_len == 0 or out_len == 0 or name_len > out_len:
return 0
pos: t.CSizeT = 0
while pos + name_len <= out_len:
if string.strncmp(out_buf + pos, global_name, name_len) == 0:
# 确认是行首(前一个是换行或 pos==0
if pos == 0 or out_buf[pos - 1] == '\n':
out_buf[pos] = ';'
return 1
pos += 1
return 0
# ============================================================
# _CommentOutOpaqueTypeInBuf - 将 out_buf 中匹配类型名的 opaque 声明行注释掉
#
# 解决跨模块类型定义冲突:本地 stub 可能包含自动生成的 opaque 声明
#(如 %"sha1.Widget" = type opaque而依赖 stub 包含完整定义
#(如 %"sha1.Widget" = type { i8*, i32 })。去重时需要将 opaque 声明
# 注释掉,让完整定义被追加,否则 LLC 报错 "Cannot allocate unsized type"。
# ============================================================
def _CommentOutOpaqueTypeInBuf(out_buf: bytes, type_name_prefix: str) -> int:
"""在 out_buf 中搜索 type_name_prefix 对应的 opaque 声明行,将行首改为 ';'
type_name_prefix 格式: '%"sha1.ClassName"'(不含 '=' 和后续内容)
返回 1=已注释, 0=未找到
"""
if out_buf is None or type_name_prefix is None:
return 0
prefix_len: t.CSizeT = string.strlen(type_name_prefix)
out_len: t.CSizeT = string.strlen(out_buf)
if prefix_len == 0 or out_len == 0 or prefix_len > out_len:
return 0
pos: t.CSizeT = 0
while pos + prefix_len <= out_len:
if string.strncmp(out_buf + pos, type_name_prefix, prefix_len) == 0:
# 确认是行首(前一个是换行或 pos==0
if pos == 0 or out_buf[pos - 1] == '\n':
# 检查后面是否是 " = type opaque"
p: t.CSizeT = pos + prefix_len
# 跳过空格
while p < out_len and out_buf[p] == ' ':
p += 1
if p < out_len and out_buf[p] == '=':
p += 1
while p < out_len and out_buf[p] == ' ':
p += 1
if p + 11 <= out_len:
if string.strncmp(out_buf + p, "type opaque", 11) == 0:
# 找到 opaque 声明,将行首改为 ';'
out_buf[pos] = ';'
return 1
pos += 1
return 0
# ============================================================
# _HasFullTypeDefinition - 检查 out_buf 中是否已有完整类型定义(非 opaque
#
# type_name_prefix 格式: '%"sha1.ClassName"'(不含 '='
# 完整定义: %"name" = type { ... } 或 %"name" = type < { ... } >
# 返回 1=已有完整定义, 0=无完整定义
# ============================================================
def _HasFullTypeDefinition(out_buf: bytes, type_name_prefix: str) -> int:
"""检查 out_buf 中是否已有 type_name_prefix 的完整定义"""
if out_buf is None or type_name_prefix is None:
return 0
prefix_len: t.CSizeT = string.strlen(type_name_prefix)
out_len: t.CSizeT = string.strlen(out_buf)
if prefix_len == 0 or out_len == 0 or prefix_len > out_len:
return 0
pos: t.CSizeT = 0
while pos + prefix_len <= out_len:
if string.strncmp(out_buf + pos, type_name_prefix, prefix_len) == 0:
# 确认是行首(前一个是换行或 pos==0且未被注释行首不是 ';'
is_line_start: int = 0
if pos == 0:
is_line_start = 1
elif out_buf[pos - 1] == '\n':
is_line_start = 1
if is_line_start == 1:
# 检查后面是否是 " = type {" 或 " = type <"
p: t.CSizeT = pos + prefix_len
while p < out_len and out_buf[p] == ' ':
p += 1
if p < out_len and out_buf[p] == '=':
p += 1
while p < out_len and out_buf[p] == ' ':
p += 1
if p + 5 < out_len:
if string.strncmp(out_buf + p, "type ", 5) == 0:
p += 5
if p < out_len:
if out_buf[p] == '{' or out_buf[p] == '<':
return 1
pos += 1
return 0
# ============================================================
# _LoadAndAppendStub - 加载依赖 stub 并追加到 out_buf
#
# 读取 {temp_dir}/{sha1}.stub.ll跳过 header;, target, source_filename
# 去重 declare/global追加到 out_buf。返回新 out_pos。
# ============================================================
def _LoadAndAppendStub(temp_dir: str, td_len: t.CSizeT, dep_sha1: str,
dep_buf: bytes, out_buf: bytes, out_size: t.CSizeT,
out_pos: t.CSizeT) -> t.CSizeT:
"""加载依赖 stub 并追加(跳过 header去重"""
if temp_dir is None or dep_sha1 is None or dep_buf is None or out_buf is None:
return out_pos
sha1_len: t.CSizeT = string.strlen(dep_sha1)
dep_path: bytes = stdlib.malloc(td_len + sha1_len + 16)
if dep_path is None:
return out_pos
viperlib.snprintf(dep_path, td_len + sha1_len + 16, "%s/%s.stub.ll", temp_dir, dep_sha1)
df_ls: fileio.File | t.CPtr = fileio.File(dep_path, fileio.MODE.R)
if df_ls.closed:
stdlib.free(dep_path)
return out_pos
dep_br_ls: t.CInt64T = df_ls.read_all(dep_buf, STUB_READ_BUF_SIZE)
df_ls.close()
stdlib.free(dep_path)
if dep_br_ls <= 0:
return out_pos
if dep_br_ls < STUB_READ_BUF_SIZE:
dep_buf[dep_br_ls] = '\0'
else:
dep_buf[STUB_READ_BUF_SIZE - 1] = '\0'
# 注释掉依赖 stub 中的字符串常量行(模块内部的,不需要)
# 简单方案:搜索 "= external unnamed_addr constant" 并将行首的 @ 改为 ;@
cfs_pos: t.CSizeT = 0
while cfs_pos + 30 < t.CSizeT(dep_br_ls):
ext_p: t.CPtr = string.strstr(dep_buf + cfs_pos, "= external unnamed_addr constant")
if ext_p is None:
break
ext_off2: t.CSizeT = t.CSizeT(t.CUInt64T(ext_p) - t.CUInt64T(dep_buf))
# 向前查找行首的 @
ls: t.CSizeT = ext_off2
while ls > 0:
if dep_buf[ls] == '\n':
ls += 1
break
ls -= 1
if ls < ext_off2 and dep_buf[ls] == '@':
dep_buf[ls] = ';'
cfs_pos = ext_off2 + 30
# 跳过 header追加内容去重 declare/global
dep_pos: t.CSizeT = 0
skipping: int = 1
while dep_pos < dep_br_ls:
line_start: t.CSizeT = dep_pos
while dep_pos < dep_br_ls:
if dep_buf[dep_pos] == '\n':
break
dep_pos += 1
line_len: t.CSizeT = dep_pos - line_start
if dep_pos < dep_br_ls:
dep_pos += 1
if skipping != 0:
if line_len == 0:
continue
ch0: int = dep_buf[line_start]
if ch0 == ';':
continue
if ch0 == 't' and line_len >= 6:
if string.strncmp(dep_buf + line_start, "target", 6) == 0:
continue
if ch0 == 's' and line_len >= 15:
if string.strncmp(dep_buf + line_start, "source_filename", 15) == 0:
continue
skipping = 0
# 去重declare用 _IsFuncDeclaredOrDefined 避免误匹配 call 语句)
if line_len >= 8 and string.strncmp(dep_buf + line_start, "declare ", 8) == 0:
at: t.CSizeT = line_start + 8
while at < line_start + line_len:
if dep_buf[at] == '@':
break
at += 1
if at < line_start + line_len:
lp: t.CSizeT = at + 1
while lp < line_start + line_len:
if dep_buf[lp] == '(':
break
lp += 1
if lp < line_start + line_len:
saved: t.CChar = dep_buf[lp + 1]
dep_buf[lp + 1] = '\0'
if _IsFuncDeclaredOrDefined(out_buf, dep_buf + at) != 0:
dep_buf[lp + 1] = saved
continue
dep_buf[lp + 1] = saved
# 去重external global
if line_len > 0 and dep_buf[line_start] == '@':
eq: t.CSizeT = line_start + 1
while eq < line_start + line_len:
if dep_buf[eq] == '=':
break
eq += 1
if eq < line_start + line_len:
saved: t.CChar = dep_buf[eq + 1]
dep_buf[eq + 1] = '\0'
if string.strstr(out_buf, dep_buf + line_start) is not None:
dep_buf[eq + 1] = saved
continue
dep_buf[eq + 1] = saved
# 去重type 定义(如 %"sha1.Point" = type {i32, i32}
# 区分完整定义(= type { ... } / = type < ... >)和 opaque 声明(= type opaque
# 当本地 stub 有 opaque 声明而依赖 stub 有完整定义时,注释掉 opaque 行,
# 让完整定义被追加,避免 LLC "Cannot allocate unsized type" 错误。
if line_len > 0 and dep_buf[line_start] == '%':
eq2: t.CSizeT = line_start + 1
while eq2 < line_start + line_len:
if dep_buf[eq2] == '=':
break
eq2 += 1
if eq2 < line_start + line_len:
saved2: t.CChar = dep_buf[eq2 + 1]
dep_buf[eq2 + 1] = '\0'
# 检查依赖行是否为完整定义(= type { 或 = type <
dep_is_full_def: int = 0
if eq2 + 8 <= line_start + line_len:
if dep_buf[eq2 + 2] == 't' and dep_buf[eq2 + 3] == 'y' and dep_buf[eq2 + 4] == 'p' and dep_buf[eq2 + 5] == 'e' and dep_buf[eq2 + 6] == ' ':
if dep_buf[eq2 + 7] == '{' or dep_buf[eq2 + 7] == '<':
dep_is_full_def = 1
if dep_is_full_def == 1:
# 依赖行是完整定义
# 临时截断到 '=' 之前(不含 '='),辅助函数期望纯类型名前缀
saved_eq: t.CChar = dep_buf[eq2]
dep_buf[eq2] = '\0'
# 1. 注释掉 out_buf 中的 opaque 声明(如果有)
commented: int = _CommentOutOpaqueTypeInBuf(out_buf, dep_buf + line_start)
# 2. 检查 out_buf 中是否已有完整定义(避免重复)
has_full: int = _HasFullTypeDefinition(out_buf, dep_buf + line_start)
dep_buf[eq2] = saved_eq
if has_full == 1:
dep_buf[eq2 + 1] = saved2
continue
# 3. 追加完整定义
else:
# 依赖行是 opaque 声明,用原有子串匹配去重
found_opaque: t.CPtr = string.strstr(out_buf, dep_buf + line_start)
if found_opaque is not None:
dep_buf[eq2 + 1] = saved2
continue
dep_buf[eq2 + 1] = saved2
# 追加
if out_pos + line_len + 2 < out_size:
string.strncpy(out_buf + out_pos, dep_buf + line_start, line_len)
out_pos += line_len
out_buf[out_pos] = '\n'
out_pos += 1
out_buf[out_pos] = '\0'
return out_pos
# ============================================================
# _LoadAndAppendTextDeclares - 从依赖 text.ll 提取 declare 追加到 out_buf
#
# 读取 {temp_dir}/{sha1}.text.ll将 define 行转为 declare提取函数签名
# 跳过函数体,去重后追加到 out_buf。返回新 out_pos。
# ============================================================
def _LoadAndAppendTextDeclares(temp_dir: str, td_len: t.CSizeT, dep_sha1: str,
dep_buf: bytes, out_buf: bytes, out_size: t.CSizeT,
out_pos: t.CSizeT) -> t.CSizeT:
"""从依赖 text.ll 提取 declaredefine 转 declare追加到 out_buf"""
if temp_dir is None or dep_sha1 is None or dep_buf is None or out_buf is None:
return out_pos
sha1_len_lt: t.CSizeT = string.strlen(dep_sha1)
text_path: bytes = stdlib.malloc(td_len + sha1_len_lt + 16)
if text_path is None:
return out_pos
viperlib.snprintf(text_path, td_len + sha1_len_lt + 16, "%s/%s.text.ll", temp_dir, dep_sha1)
tf_lt: fileio.File | t.CPtr = fileio.File(text_path, fileio.MODE.R)
if tf_lt.closed:
stdlib.free(text_path)
return out_pos
text_br: t.CInt64T = tf_lt.read_all(dep_buf, STUB_READ_BUF_SIZE)
tf_lt.close()
stdlib.free(text_path)
if text_br <= 0:
return out_pos
if text_br < STUB_READ_BUF_SIZE:
dep_buf[text_br] = '\0'
else:
dep_buf[STUB_READ_BUF_SIZE - 1] = '\0'
dep_pos_lt: t.CSizeT = 0
in_body: int = 0
while dep_pos_lt < text_br:
ls: t.CSizeT = dep_pos_lt
while dep_pos_lt < text_br:
if dep_buf[dep_pos_lt] == '\n':
break
dep_pos_lt += 1
ll: t.CSizeT = dep_pos_lt - ls
if dep_pos_lt < text_br:
dep_pos_lt += 1
# 函数体跳过:直到单独的 } 行
if in_body != 0:
if ll == 1 and dep_buf[ls] == '}':
in_body = 0
continue
# 跳过空行和 header
if ll == 0:
continue
c0: int = dep_buf[ls]
if c0 == ';':
continue
if c0 == 't' and ll >= 6 and string.strncmp(dep_buf + ls, "target", 6) == 0:
continue
if c0 == 's' and ll >= 15 and string.strncmp(dep_buf + ls, "source_filename", 15) == 0:
continue
# 处理 define 行:转为 declare
if ll >= 7 and string.strncmp(dep_buf + ls, "define ", 7) == 0:
# 提取 @funcname( 用于去重
at_lt: t.CSizeT = ls + 7
while at_lt < ls + ll:
if dep_buf[at_lt] == '@':
break
at_lt += 1
if at_lt >= ls + ll:
in_body = 1
continue
lp_lt: t.CSizeT = at_lt + 1
while lp_lt < ls + ll:
if dep_buf[lp_lt] == '(':
break
lp_lt += 1
if lp_lt >= ls + ll:
in_body = 1
continue
# 去重检查:用 _IsFuncDeclaredOrDefined 避免误匹配 call 语句
saved_lt: t.CChar = dep_buf[lp_lt + 1]
dep_buf[lp_lt + 1] = '\0'
if _IsFuncDeclaredOrDefined(out_buf, dep_buf + at_lt) != 0:
dep_buf[lp_lt + 1] = saved_lt
in_body = 1
continue
dep_buf[lp_lt + 1] = saved_lt
# 找行尾的 ' {' 并截断
decl_end: t.CSizeT = ls + ll
if decl_end > ls + 1:
if dep_buf[decl_end - 1] == '{' and dep_buf[decl_end - 2] == ' ':
decl_end -= 2
# 输出: declare + content (skip "define " 7 chars, up to decl_end)
copy_len_lt: t.CSizeT = decl_end - (ls + 7)
if out_pos + 8 + copy_len_lt + 2 < out_size:
string.strncpy(out_buf + out_pos, "declare ", 8)
out_pos += 8
string.strncpy(out_buf + out_pos, dep_buf + ls + 7, copy_len_lt)
out_pos += copy_len_lt
out_buf[out_pos] = '\n'
out_pos += 1
out_buf[out_pos] = '\0'
in_body = 1
continue
# 跳过其他行global 带初值定义、type 定义等stub 已有 external 声明)
return out_pos
# ============================================================
# _PathToModuleName - 将文件路径转换为模块名
#
# 输入: "includes/stdio.py" 或 "includes/lib/core/Handles/HandlesType.py"
# 输出: "stdio" 或 "lib.core.Handles.HandlesType"
# 去除 includes/ 前缀和 .py 后缀,/ 替换为 .
# 返回 stdlib.malloc 分配的字符串,调用者负责释放
# ============================================================
def _PathToModuleName(path: str) -> str:
"""将文件路径转换为模块名
特殊处理 __init__.pyincludes/ast/__init__.py → ast而非 ast.__init__
"""
if path is None:
return None
plen: t.CSizeT = string.strlen(path)
# 跳过 includes/ 前缀
prefix: str = "includes/"
prefix_len: t.CSizeT = 9
path_start: t.CSizeT = 0
if plen > prefix_len and string.strncmp(path, prefix, prefix_len) == 0:
path_start = prefix_len
plen = plen - prefix_len
# 去掉 .py 后缀
if plen > 3 and path[path_start + plen - 3] == '.' and path[path_start + plen - 2] == 'p' and path[path_start + plen - 1] == 'y':
plen -= 3
if plen == 0:
return None
# 检查是否以 /__init__ 或 \__init__ 结尾(包的 __init__.py
# /__init__ 是 9 个字符:/ _ _ i n i t _ _
# 如果是,去掉分隔符+__init__ 后缀,模块名就是包名(如 ast/__init__ → ast
# 注意:顶级 __init__.pyplen=8不会进入此分支因为 plen > 9 才检查
# 同时支持 / 和 \ 两种分隔符TPC 写入的路径可能混合使用)
if plen > 9:
tail_off: t.CSizeT = path_start + plen - 9
sep_char: t.CChar = path[tail_off]
if (sep_char == '/' or sep_char == '\\') \
and path[tail_off + 1] == '_' and path[tail_off + 2] == '_' \
and path[tail_off + 3] == 'i' and path[tail_off + 4] == 'n' and path[tail_off + 5] == 'i' \
and path[tail_off + 6] == 't' and path[tail_off + 7] == '_' and path[tail_off + 8] == '_':
# 去掉 分隔符+__init__ 后缀plen 就是包名部分长度
plen = tail_off - path_start
if plen == 0:
return None
# 复制并替换 / 为 .
result: str = stdlib.malloc(plen + 1)
if result is None:
return None
for ri in range(plen):
ch: t.CChar = path[path_start + ri]
if ch == '/' or ch == '\\':
result[ri] = '.'
else:
result[ri] = ch
result[plen] = '\0'
return result
# ============================================================
# _BuildIncludesSha1Map - 从 _sha1_map.txt 构建 sha1 → module_name 映射
#
# _sha1_map.txt 格式: {sha1}:includes/{rel_path}\n
# 返回两个数组sha1_list 和 mod_list长度存储在 count 中。
# 所有内存由 stdlib.malloc 分配,调用者负责释放。
# ============================================================
def _BuildIncludesSha1Map(temp_dir: str, td_len: t.CSizeT,
sha1_list: t.CChar | t.CPtr,
mod_list: t.CChar | t.CPtr) -> int:
"""从 _sha1_map.txt 构建 sha1→module_name 映射,返回条目数"""
if temp_dir is None or sha1_list is None or mod_list is None:
return 0
map_path: bytes = stdlib.malloc(td_len + 32)
if map_path is None:
return 0
viperlib.snprintf(map_path, td_len + 32, "%s/_sha1_map.txt", temp_dir)
mf: fileio.File | t.CPtr = fileio.File(map_path, fileio.MODE.R)
if mf.closed:
stdlib.free(map_path)
return 0
map_buf: bytes = stdlib.malloc(STUB_READ_BUF_SIZE)
if map_buf is None:
mf.close()
stdlib.free(map_path)
return 0
map_br: t.CInt64T = mf.read_all(map_buf, STUB_READ_BUF_SIZE)
mf.close()
stdlib.free(map_path)
if map_br <= 0:
stdlib.free(map_buf)
return 0
if map_br < STUB_READ_BUF_SIZE:
map_buf[map_br] = '\0'
else:
map_buf[STUB_READ_BUF_SIZE - 1] = '\0'
stdio.printf("[BISM] map_br=%lld\n", map_br)
stdio.fflush(0)
# 堆探针:测试堆是否已被之前的操作损坏
probe: bytes = stdlib.malloc(17)
stdio.printf("[BISM] probe malloc(17)=%p\n", probe)
stdio.fflush(0)
if probe is not None:
stdlib.free(probe)
stdio.printf("[BISM] probe free OK\n")
stdio.fflush(0)
count: int = 0
pos: t.CSizeT = 0
while pos < map_br and count < MAX_INCLUDES:
stdio.printf("[BISM] iter=%d pos=%d\n", count, pos)
stdio.fflush(0)
# 提取 SHA116 hex
sha1: t.CChar | t.CPtr = stdlib.malloc(17)
stdio.printf("[BISM] malloc(17)=%p\n", sha1)
stdio.fflush(0)
if sha1 is None:
break
hv: int = win32memory.HeapValidate(win32memory.GetProcessHeap(), 0, None)
stdio.printf("[BISM] HV after malloc(sha1)=%d\n", hv)
stdio.fflush(0)
string.strncpy(sha1, map_buf + pos, 16)
sha1[16] = '\0'
pos += 16
# 跳过 ':'
if pos < map_br and map_buf[pos] == ':':
pos += 1
# 提取路径(到 \r、\n 或 \0
# 注意: Windows CRLF 换行符是 \r\n必须同时检查 \r 避免路径包含 \r
path_start: t.CSizeT = pos
while pos < map_br and map_buf[pos] != '\n' and map_buf[pos] != '\r' and map_buf[pos] != '\0':
pos += 1
path_len: t.CSizeT = pos - path_start
stdio.printf("[BISM] path_len=%d\n", path_len)
stdio.fflush(0)
# 跳过 \r\n 或 \n
if pos < map_br and map_buf[pos] == '\r':
pos += 1
if pos < map_br and map_buf[pos] == '\n':
pos += 1
if path_len > 0:
# 仅收集 includes/ 前缀的条目(用户文件如 App/ 不应进入 includes 映射)
# 否则用户文件会被误判为 includes导致 is_module_imported 模块名不匹配而跳过加载
if path_len >= 9 and string.strncmp(map_buf + path_start, "includes/", 9) == 0:
path_buf: t.CChar | t.CPtr = stdlib.malloc(path_len + 1)
stdio.printf("[BISM] malloc(%d)=%p\n", path_len + 1, path_buf)
stdio.fflush(0)
if path_buf is not None:
string.strncpy(path_buf, map_buf + path_start, path_len)
path_buf[path_len] = '\0'
hv2: int = win32memory.HeapValidate(win32memory.GetProcessHeap(), 0, None)
stdio.printf("[BISM] HV after fill(path_buf)=%d\n", hv2)
stdio.fflush(0)
mod_name: str = _PathToModuleName(path_buf)
stdio.printf("[BISM] _PathToModuleName=%p\n", mod_name)
stdio.fflush(0)
hv3: int = win32memory.HeapValidate(win32memory.GetProcessHeap(), 0, None)
stdio.printf("[BISM] HV after _PathToModuleName=%d\n", hv3)
stdio.fflush(0)
stdlib.free(path_buf)
if mod_name is not None:
# 存储 sha1 和 mod_name 到数组中
idx: t.CSizeT = t.CSizeT(count) * 17
string.strcpy(sha1_list + idx, sha1)
hv4: int = win32memory.HeapValidate(win32memory.GetProcessHeap(), 0, None)
stdio.printf("[BISM] HV after strcpy(sha1_list)=%d\n", hv4)
stdio.fflush(0)
idx2: t.CSizeT = t.CSizeT(count) * 64
mn_len: t.CSizeT = string.strlen(mod_name)
stdio.printf("[BISM] mn_len=%d count=%d\n", mn_len, count)
stdio.fflush(0)
if mn_len < 64:
string.strcpy(mod_list + idx2, mod_name)
else:
string.strncpy(mod_list + idx2, mod_name, 63)
mod_list[idx2 + 63] = '\0'
hv5: int = win32memory.HeapValidate(win32memory.GetProcessHeap(), 0, None)
stdio.printf("[BISM] HV after strcpy(mod_list)=%d\n", hv5)
stdio.fflush(0)
stdlib.free(mod_name)
stdio.printf("[BISM] free(mod_name) OK\n")
stdio.fflush(0)
count += 1
stdio.printf("[BISM] before free(sha1=%p)\n", sha1)
stdio.fflush(0)
stdlib.free(sha1)
stdio.printf("[BISM] free(sha1) OK\n")
stdio.fflush(0)
stdlib.free(map_buf)
return count
# ============================================================
# _FindSha1ByModName - 从模块名查找 SHA1精确匹配
#
# 在 sha1_arr/mod_arr 映射中查找模块名对应的 SHA1。
# 用于依赖图分析:模块名 → SHA1。
#
# Args:
# sha1_arr: SHA1 数组(每个 17 字节)
# mod_arr: 模块名数组(每个 64 字节)
# count: 映射条目数
# mod_name: 要查找的模块名
#
# Returns:
# SHA1 字符串指针(指向 sha1_arr 内部,无需释放),未找到返回 None
# ============================================================
def _FindSha1ByModName(sha1_arr: t.CChar | t.CPtr, mod_arr: t.CChar | t.CPtr,
count: int, mod_name: str) -> str:
"""从模块名查找 SHA1精确匹配"""
if sha1_arr is None or mod_arr is None or mod_name is None:
return None
if count <= 0:
return None
for i in range(count):
idx: t.CSizeT = t.CSizeT(i) * 64
if string.strcmp(mod_arr + idx, mod_name) == 0:
sidx: t.CSizeT = t.CSizeT(i) * 17
return sha1_arr + sidx
return None
# ============================================================
# _AddSha1ToSet - 将 SHA1 加入集合(去重)
#
# 检查 SHA1 是否已在集合中,若不在则加入。返回新的 count。
#
# Args:
# set_buf: SHA1 集合缓冲区(每个 17 字节)
# count: 当前集合中的 SHA1 数量
# sha1: 要加入的 SHA1
#
# Returns:
# 新的 count如果已存在则不变
# ============================================================
def _AddSha1ToSet(set_buf: t.CChar | t.CPtr, count: int, sha1: str) -> int:
"""将 SHA1 加入集合,返回新 count"""
if set_buf is None or sha1 is None:
return count
for i in range(count):
idx: t.CSizeT = t.CSizeT(i) * 17
if string.strcmp(set_buf + idx, sha1) == 0:
return count
if count < MAX_INCLUDES_SHA1:
idx: t.CSizeT = t.CSizeT(count) * 17
string.strcpy(set_buf + idx, sha1)
return count + 1
return count
# ============================================================
# _BuildReachableSha1Set - 构建可达 SHA1 集合(依赖图按需翻译)
#
# 从 source_dir 的源文件开始,解析 import 语句,递归收集可达的
# includes 文件 SHA1。Phase 1b 只翻译可达集合中的文件,避免
# 翻译不需要的 includes如 Test 不依赖 llvmlite则不翻译
#
# 算法(工作列表):
# 1. 扫描 source_dir 下的 .py 文件,解析 AST 获取 _imported_modules
# 2. 构建 SHA1→module_name 映射(从 _sha1_map.txt
# 3. 工作列表递归:模块名 → 查 SHA1 → 加入 reachable_set → 读 .deps.txt → 追加新模块名
#
# Args:
# mb: 内存池
# source_dir: 源文件目录Config.SourceDir
# temp_dir: 临时目录_sha1_map.txt 和 .deps.txt 所在位置)
# reachable_set: 输出参数,可达 SHA1 集合缓冲区MAX_INCLUDES_SHA1 * 17 字节)
#
# Returns:
# 可达 SHA1 数量(>0 成功,<=0 失败)
# ============================================================
def _BuildReachableSha1Set(mb: memhub.MemBuddy | t.CPtr, source_dir: str,
temp_dir: str,
reachable_set: t.CChar | t.CPtr) -> int:
"""构建可达 SHA1 集合(依赖图按需翻译)"""
if source_dir is None or temp_dir is None or reachable_set is None:
return -1
SRC_BUF_SIZE_R: t.CSizeT = 1048576
# 1. 构建 SHA1→module_name 映射
td_len_r: t.CSizeT = string.strlen(temp_dir)
sha1_arr: bytes = stdlib.malloc(MAX_INCLUDES * 17)
mod_arr: bytes = stdlib.malloc(MAX_INCLUDES * 64)
if sha1_arr is None or mod_arr is None:
if sha1_arr is not None:
stdlib.free(sha1_arr)
if mod_arr is not None:
stdlib.free(mod_arr)
return -1
map_count: int = _BuildIncludesSha1Map(temp_dir, td_len_r, sha1_arr, mod_arr)
if map_count <= 0:
stdio.printf("[Reachable] 无法构建 SHA1 映射\n")
stdlib.free(sha1_arr)
stdlib.free(mod_arr)
return -1
stdio.printf("[Reachable] SHA1 映射: %d\n", map_count)
# 2. 扫描 source_dir 下的 .py 文件,收集直接依赖
dir_len: t.CSizeT = string.strlen(source_dir)
pattern: bytes = stdlib.malloc(dir_len + 8)
if pattern is None:
stdlib.free(sha1_arr)
stdlib.free(mod_arr)
return -1
viperlib.snprintf(pattern, dir_len + 8, "%s/*.py", source_dir)
find_data_size: t.CSizeT = win32file.WIN32_FIND_DATAA.__sizeof__()
find_data: win32file.WIN32_FIND_DATAA | t.CPtr = stdlib.malloc(find_data_size + 16)
if find_data is None:
stdlib.free(pattern)
stdlib.free(sha1_arr)
stdlib.free(mod_arr)
return -1
string.memset(find_data, 0, find_data_size + 16)
# 工作列表(空格分隔的模块名)
worklist: bytes = stdlib.malloc(8192)
if worklist is None:
stdlib.free(pattern)
stdlib.free(find_data)
stdlib.free(sha1_arr)
stdlib.free(mod_arr)
return -1
worklist[0] = '\0'
wl_len: t.CSizeT = 0
reachable_count: int = 0
handle: win32base.HANDLE = win32file.FindFirstFileA(pattern, find_data)
if handle == win32base.INVALID_HANDLE_VALUE:
stdio.printf("[Reachable] 未找到 .py 文件: %s\n", pattern)
stdlib.free(pattern)
stdlib.free(find_data)
stdlib.free(sha1_arr)
stdlib.free(mod_arr)
stdlib.free(worklist)
return -1
while True:
fname: str = find_data.cFileName
if fname is not None:
fname_len: t.CSizeT = string.strlen(fname)
if fname_len > 3:
is_py: int = 0
if fname[fname_len - 3] == '.' and fname[fname_len - 2] == 'p' and fname[fname_len - 1] == 'y':
is_py = 1
if is_py != 0:
full_path: bytes = stdlib.malloc(dir_len + fname_len + 2)
if full_path is not None:
viperlib.snprintf(full_path, dir_len + fname_len + 2, "%s/%s", source_dir, fname)
sf: fileio.File | t.CPtr = fileio.File(full_path, fileio.MODE.R)
if not sf.closed:
sbuf: bytes = stdlib.malloc(SRC_BUF_SIZE_R)
if sbuf is not None:
br: LONG = sf.read_all(sbuf, SRC_BUF_SIZE_R)
sf.close()
if br > 0:
if br < SRC_BUF_SIZE_R:
sbuf[br] = 0
else:
sbuf[SRC_BUF_SIZE_R - 1] = 0
# 解析 AST 获取 _imported_modules
lx: ast.Lexer | t.CPtr = ast.new_lexer(mb)
if lx is not None:
ast._lexer_init(lx, sbuf, mb)
tokens: ast.Token | t.CPtr = ast.tokenize(lx)
tree: ast.AST | t.CPtr = ast.parse_tokens(mb, tokens)
if tree is not None:
tr: HandlesTranslator.Translator | t.CPtr = HandlesTranslator.Translator()
if tr is not None:
tr._declare_only = 2
# 源文件包名fname 是顶级文件名(无目录分隔符),包为 None
tr.CurrentPackage = HandlesImports.compute_package_from_relpath(mb, fname)
HandlesStruct.reset_visible_structs(mb, 0)
tr.translate(tree)
# 获取 _imported_modules
if tr._imported_modules is not None:
im_len: t.CSizeT = string.strlen(tr._imported_modules)
if im_len > 0 and wl_len + im_len + 1 < 8192:
string.strcpy(worklist + wl_len, tr._imported_modules)
wl_len += im_len
worklist[wl_len] = ' '
wl_len += 1
worklist[wl_len] = '\0'
# 释放 Translator 资源
if tr._global_names is not None:
stdlib.free(tr._global_names)
if tr._nonlocal_names is not None:
stdlib.free(tr._nonlocal_names)
stdlib.free(sbuf)
else:
stdlib.free(sbuf)
else:
sf.close()
stdlib.free(full_path)
if win32file.FindNextFileA(handle, find_data) == 0:
break
win32base.FindClose(handle)
stdlib.free(pattern)
stdlib.free(find_data)
# 去掉末尾多余空格
if wl_len > 0 and worklist[wl_len - 1] == ' ':
worklist[wl_len - 1] = '\0'
wl_len -= 1
stdio.printf("[Reachable] 源文件直接依赖: '%s'\n", worklist)
# 2.5. 全局预加入基础容器模块_list/_dict/json
#
# 这些模块定义了 list/dict 等内建容器的泛型模板。
# 使用 list[...]/dict[...] 语法时需要这些模块被翻译并注册模板。
# 即使源文件和 includes 文件没有显式 import _list也预加入以确保可达。
# 这对应 TPC 的 _inject_auto_imports 机制TPC 在翻译前自动注入 import _list
# _list
if HandlesImports.is_module_imported(worklist, "_list") == 0:
bsha1_l: str = _FindSha1ByModName(sha1_arr, mod_arr, map_count, "_list")
if bsha1_l is not None:
if wl_len + 7 < 8192:
if wl_len > 0:
worklist[wl_len] = ' '
wl_len += 1
string.strcpy(worklist + wl_len, "_list")
wl_len += 5
worklist[wl_len] = ' '
wl_len += 1
worklist[wl_len] = '\0'
# _dict
if HandlesImports.is_module_imported(worklist, "_dict") == 0:
bsha1_d: str = _FindSha1ByModName(sha1_arr, mod_arr, map_count, "_dict")
if bsha1_d is not None:
if wl_len + 7 < 8192:
if wl_len > 0:
worklist[wl_len] = ' '
wl_len += 1
string.strcpy(worklist + wl_len, "_dict")
wl_len += 5
worklist[wl_len] = ' '
wl_len += 1
worklist[wl_len] = '\0'
# json
if HandlesImports.is_module_imported(worklist, "json") == 0:
bsha1_j: str = _FindSha1ByModName(sha1_arr, mod_arr, map_count, "json")
if bsha1_j is not None:
if wl_len + 6 < 8192:
if wl_len > 0:
worklist[wl_len] = ' '
wl_len += 1
string.strcpy(worklist + wl_len, "json")
wl_len += 4
worklist[wl_len] = ' '
wl_len += 1
worklist[wl_len] = '\0'
stdio.printf("[Reachable] 预加入容器模块后: '%s'\n", worklist)
# 3. 工作列表算法:递归收集可达 SHA1
processed: bytes = stdlib.malloc(8192)
if processed is None:
stdlib.free(sha1_arr)
stdlib.free(mod_arr)
stdlib.free(worklist)
return -1
processed[0] = '\0'
proc_len: t.CSizeT = 0
wl_pos: t.CSizeT = 0
wl_iter: int = 0
while wl_pos < wl_len:
wl_iter += 1
# 提取一个模块名(到空格或末尾)
name_start: t.CSizeT = wl_pos
while wl_pos < wl_len and worklist[wl_pos] != ' ':
wl_pos += 1
name_len: t.CSizeT = wl_pos - name_start
if wl_pos < wl_len:
wl_pos += 1
if name_len == 0:
continue
# 复制模块名
mod_buf: bytes = stdlib.malloc(name_len + 1)
if mod_buf is None:
continue
string.strncpy(mod_buf, worklist + name_start, name_len)
mod_buf[name_len] = '\0'
# 检查是否已处理
already: int = 0
if HandlesImports.is_module_imported(processed, mod_buf) != 0:
already = 1
if already == 0:
# 加入 processed
if proc_len + name_len + 1 < 8192:
string.strcpy(processed + proc_len, mod_buf)
proc_len += name_len
processed[proc_len] = ' '
proc_len += 1
processed[proc_len] = '\0'
# 查找 SHA1
found_sha1: str = _FindSha1ByModName(sha1_arr, mod_arr, map_count, mod_buf)
if found_sha1 is not None:
# 加入 reachable_set
reachable_count = _AddSha1ToSet(reachable_set, reachable_count, found_sha1)
# 读取 .deps.txt 追加到 worklist
deps_path: bytes = stdlib.malloc(td_len_r + 33)
if deps_path is not None:
viperlib.snprintf(deps_path, td_len_r + 33, "%s/%s.deps.txt", temp_dir, found_sha1)
df: fileio.File | t.CPtr = fileio.File(deps_path, fileio.MODE.R)
if not df.closed:
deps_buf: bytes = stdlib.malloc(2048)
if deps_buf is not None:
dbr: t.CInt64T = df.read_all(deps_buf, 2048)
df.close()
if dbr > 0:
if dbr < 2048:
deps_buf[dbr] = '\0'
else:
deps_buf[2047] = '\0'
# 追加到 worklist确保前面有空格分隔符避免模块名合并
dl: t.CSizeT = string.strlen(deps_buf)
if dl > 0 and wl_len + dl + 2 < 8192:
if wl_len > 0 and worklist[wl_len - 1] != ' ':
worklist[wl_len] = ' '
wl_len += 1
string.strcpy(worklist + wl_len, deps_buf)
wl_len += dl
worklist[wl_len] = ' '
wl_len += 1
worklist[wl_len] = '\0'
stdlib.free(deps_buf)
else:
df.close()
stdlib.free(deps_path)
stdlib.free(mod_buf)
stdlib.free(sha1_arr)
stdlib.free(mod_arr)
stdlib.free(worklist)
stdlib.free(processed)
stdio.printf("[Reachable] 可达 SHA1: %d\n", reachable_count)
return reachable_count
# ============================================================
# BuildCombinedIR - 组合本地 stub + 依赖 stubs + 本地 text → 完整 IR
#
# 按需加载:读取 deps.txt 获取导入模块名,通过 _sha1_map.txt 查找 SHA1
# 仅加载实际使用的依赖 stub而非全部扫描
# ============================================================
def BuildCombinedIR(temp_dir: str, local_sha1: str,
out_buf: bytes, out_size: t.CSizeT) -> t.CSizeT:
"""组合本地 stub + 依赖 stubs + 本地 text → out_buf按需加载"""
if temp_dir is None or local_sha1 is None or out_buf is None or out_size == 0:
return 0
stdio.printf("[BuildCombinedIR] start: sha1=%s\n", local_sha1)
td_len: t.CSizeT = string.strlen(temp_dir)
out_buf[0] = '\0'
out_pos: t.CSizeT = 0
# 1. 读取并追加本地 stub.ll含 header
stub_path: bytes = stdlib.malloc(td_len + 32)
if stub_path is None:
return 0
viperlib.snprintf(stub_path, td_len + 32, "%s/%s.stub.ll", temp_dir, local_sha1)
sf: fileio.File | t.CPtr = fileio.File(stub_path, fileio.MODE.R)
if sf.closed:
stdlib.free(stub_path)
return 0
stub_content: bytes = stdlib.malloc(STUB_READ_BUF_SIZE)
if stub_content is None:
sf.close()
stdlib.free(stub_path)
return 0
stub_br: t.CInt64T = sf.read_all(stub_content, STUB_READ_BUF_SIZE)
sf.close()
stdlib.free(stub_path)
stdio.printf("[BuildCombinedIR] stub read: %d bytes\n", stub_br)
if stub_br <= 0:
stdlib.free(stub_content)
return 0
if stub_br < STUB_READ_BUF_SIZE:
stub_content[stub_br] = '\0'
else:
stub_content[STUB_READ_BUF_SIZE - 1] = '\0'
slen: t.CSizeT = stub_br
# 修复本地 stub.ll 中字符串常量的非法 IR
# llvmlite OUTPUT_STUB 模式输出 "= external unnamed_addr constant" 的非法形式。
# 简单方案:搜索 "= external unnamed_addr constant" 并将 external 替换为 8 个空格。
fix_pos: t.CSizeT = 0
while fix_pos + 30 < slen:
ext_ptr: t.CPtr = string.strstr(stub_content + fix_pos, "= external unnamed_addr constant")
if ext_ptr is None:
break
# ext_ptr 指向 "= external ..." 中的 '='
# 'external' 开始于 ext_ptr + 2
ext_off: t.CSizeT = t.CSizeT(t.CUInt64T(ext_ptr) - t.CUInt64T(stub_content)) + 2
if ext_off + 8 <= slen:
stub_content[ext_off] = ' '
stub_content[ext_off + 1] = ' '
stub_content[ext_off + 2] = ' '
stub_content[ext_off + 3] = ' '
stub_content[ext_off + 4] = ' '
stub_content[ext_off + 5] = ' '
stub_content[ext_off + 6] = ' '
stub_content[ext_off + 7] = ' '
fix_pos = ext_off + 8
stdio.printf("[BuildCombinedIR] stub_fix done, slen=%d\n", slen)
# 直接复制修复后的 stub 内容
if out_pos + slen + 2 < out_size:
string.strcpy(out_buf + out_pos, stub_content)
out_pos += slen
if out_buf[out_pos - 1] != '\n':
out_buf[out_pos] = '\n'
out_pos += 1
out_buf[out_pos] = '\0'
stdlib.free(stub_content)
# 2. 按需加载依赖 stubs根据 deps.txt 过滤,而非扫描全部)
# 先加载依赖 stub含 type 定义),确保 type 定义在本地 text.ll 的 define 块之前
# 2a. 构建 includes SHA1 → module_name 映射(从 _sha1_map.txt
sha1_arr: bytes = stdlib.malloc(MAX_INCLUDES * 17)
mod_arr: bytes = stdlib.malloc(MAX_INCLUDES * 64)
inc_count: int = 0
if sha1_arr is not None and mod_arr is not None:
inc_count = _BuildIncludesSha1Map(temp_dir, td_len, sha1_arr, mod_arr)
# 2b. 读取 deps.txt 获取导入模块名集合
deps_buf: str = None
deps_loaded: int = 0
deps_path: bytes = stdlib.malloc(td_len + 32)
if deps_path is not None:
viperlib.snprintf(deps_path, td_len + 32, "%s/%s.deps.txt", temp_dir, local_sha1)
df_deps: fileio.File | t.CPtr = fileio.File(deps_path, fileio.MODE.R)
if not df_deps.closed:
deps_buf = stdlib.malloc(4096)
if deps_buf is not None:
deps_br: t.CInt64T = df_deps.read_all(deps_buf, 4096)
if deps_br > 0:
if deps_br < 4096:
deps_buf[deps_br] = '\0'
else:
deps_buf[4095] = '\0'
deps_loaded = 1
df_deps.close()
stdlib.free(deps_path)
# 2c. 扫描 temp_dir 中所有 .stub.ll按需加载
pattern: bytes = stdlib.malloc(td_len + 16)
if pattern is not None:
viperlib.snprintf(pattern, td_len + 16, "%s/*.stub.ll", temp_dir)
find_data: win32file.WIN32_FIND_DATAA | t.CPtr = stdlib.malloc(win32file.WIN32_FIND_DATAA.__sizeof__())
if find_data is not None:
string.memset(find_data, 0, win32file.WIN32_FIND_DATAA.__sizeof__())
handle: win32base.HANDLE = win32file.FindFirstFileA(pattern, find_data)
if handle != win32base.INVALID_HANDLE_VALUE:
dep_buf: bytes = stdlib.malloc(STUB_READ_BUF_SIZE)
if dep_buf is not None:
while 1:
fname: str = find_data.cFileName
if fname is not None:
dep_sha1: str = stdlib.malloc(17)
if dep_sha1 is not None:
string.strncpy(dep_sha1, fname, 16)
dep_sha1[16] = '\0'
if string.strcmp(dep_sha1, local_sha1) != 0:
should_load: int = 0
# 检查是否在 includes 映射中
is_include: int = 0
if inc_count > 0:
for ii in range(inc_count):
idx_ii: t.CSizeT = t.CSizeT(ii) * 17
if string.strcmp(sha1_arr + idx_ii, dep_sha1) == 0:
is_include = 1
if deps_loaded != 0:
idx_mi: t.CSizeT = t.CSizeT(ii) * 64
if HandlesImports.is_module_imported(deps_buf, mod_arr + idx_mi) != 0:
should_load = 1
break
if is_include == 0:
# 非 includes stub用户文件总是加载
should_load = 1
if should_load != 0:
out_pos = _LoadAndAppendStub(temp_dir, td_len, dep_sha1, dep_buf, out_buf, out_size, out_pos)
out_pos = _LoadAndAppendTextDeclares(temp_dir, td_len, dep_sha1, dep_buf, out_buf, out_size, out_pos)
stdlib.free(dep_sha1)
if win32file.FindNextFileA(handle, find_data) == 0:
break
if dep_buf is not None:
stdlib.free(dep_buf)
win32file.FindClose(handle)
stdlib.free(find_data)
stdlib.free(pattern)
# 2d. 验证 deps.txt 中所有依赖模块的 stub 文件都存在fail-fast
# 避免到 llc 才报 undefined value 错误。
# 仅验证在 includes 映射中找到的模块t/c 等内部库无 SHA1 映射,自动跳过)。
if deps_loaded != 0 and inc_count > 0:
dep_total: t.CSizeT = string.strlen(deps_buf)
dp: t.CSizeT = 0
while dp < dep_total:
# 跳过前导空格
while dp < dep_total:
if deps_buf[dp] != ' ':
break
dp += 1
if dp >= dep_total:
break
# 提取模块名(到下一个空格或末尾)
ms: t.CSizeT = dp
while dp < dep_total:
if deps_buf[dp] == ' ':
break
dp += 1
ml: t.CSizeT = dp - ms
if ml == 0:
continue
# 复制模块名到临时缓冲区并 null 终止
mod_nm: str = stdlib.malloc(ml + 1)
if mod_nm is None:
continue
string.strncpy(mod_nm, deps_buf + ms, ml)
mod_nm[ml] = '\0'
# 在 includes 映射中查找模块对应的 SHA1
found_s: str = None
for mi2 in range(inc_count):
idx_m2: t.CSizeT = t.CSizeT(mi2) * 64
if string.strcmp(mod_arr + idx_m2, mod_nm) == 0:
idx_s2: t.CSizeT = t.CSizeT(mi2) * 17
found_s = sha1_arr + idx_s2
break
if found_s is not None:
# 检查 stub 文件是否存在
chk_path: bytes = stdlib.malloc(td_len + 32)
if chk_path is not None:
viperlib.snprintf(chk_path, td_len + 32, "%s/%s.stub.ll", temp_dir, found_s)
chk_f: fileio.File | t.CPtr = fileio.File(chk_path, fileio.MODE.R)
if chk_f.closed:
stdio.printf("[FATAL][BuildCombinedIR] 依赖模块 '%s' (sha1=%s) 的 stub 文件不存在: %s,立即终止编译\n", mod_nm, found_s, chk_path)
sys.exit(1)
chk_f.close()
stdlib.free(chk_path)
stdlib.free(mod_nm)
# 3. 读取并追加本地 text.ll行级去重跳过 stub.ll 中已存在的定义)
text_path: bytes = stdlib.malloc(td_len + 32)
if text_path is None:
return out_pos
viperlib.snprintf(text_path, td_len + 32, "%s/%s.text.ll", temp_dir, local_sha1)
tf: fileio.File | t.CPtr = fileio.File(text_path, fileio.MODE.R)
if tf.closed:
stdlib.free(text_path)
# text.ll 不存在的情况,继续
else:
text_content: bytes = stdlib.malloc(STUB_READ_BUF_SIZE)
if text_content is None:
tf.close()
stdlib.free(text_path)
else:
text_br: t.CInt64T = tf.read_all(text_content, STUB_READ_BUF_SIZE)
tf.close()
stdlib.free(text_path)
if text_br > 0:
if text_br < STUB_READ_BUF_SIZE:
text_content[text_br] = '\0'
else:
text_content[STUB_READ_BUF_SIZE - 1] = '\0'
# 行级去重追加(与 _LoadAndAppendStub 相同逻辑)
tx_pos: t.CSizeT = 0
tx_skip: int = 1
while tx_pos < text_br:
tx_ls: t.CSizeT = tx_pos
while tx_pos < text_br:
if text_content[tx_pos] == '\n':
break
tx_pos += 1
tx_ll: t.CSizeT = tx_pos - tx_ls
if tx_pos < text_br:
tx_pos += 1
if tx_skip != 0:
if tx_ll == 0:
continue
tx_c0: int = text_content[tx_ls]
if tx_c0 == ';':
continue
if tx_c0 == 't' and tx_ll >= 6:
if string.strncmp(text_content + tx_ls, "target", 6) == 0:
continue
if tx_c0 == 's' and tx_ll >= 15:
if string.strncmp(text_content + tx_ls, "source_filename", 15) == 0:
continue
tx_skip = 0
# 去重declare用 _IsFuncDeclaredOrDefined 避免误匹配 call 语句)
if tx_ll >= 8 and string.strncmp(text_content + tx_ls, "declare ", 8) == 0:
tx_at: t.CSizeT = tx_ls + 8
while tx_at < tx_ls + tx_ll:
if text_content[tx_at] == '@':
break
tx_at += 1
if tx_at < tx_ls + tx_ll:
tx_lp: t.CSizeT = tx_at + 1
while tx_lp < tx_ls + tx_ll:
if text_content[tx_lp] == '(':
break
tx_lp += 1
if tx_lp < tx_ls + tx_ll:
tx_sv: t.CChar = text_content[tx_lp + 1]
text_content[tx_lp + 1] = '\0'
if _IsFuncDeclaredOrDefined(out_buf, text_content + tx_at) != 0:
text_content[tx_lp + 1] = tx_sv
continue
text_content[tx_lp + 1] = tx_sv
# 去重define注释掉 out_buf 中已有的 declare避免 declare/define 冲突)
if tx_ll >= 7 and string.strncmp(text_content + tx_ls, "define ", 7) == 0:
tx_at_d: t.CSizeT = tx_ls + 7
while tx_at_d < tx_ls + tx_ll:
if text_content[tx_at_d] == '@':
break
tx_at_d += 1
if tx_at_d < tx_ls + tx_ll:
tx_lp_d: t.CSizeT = tx_at_d + 1
while tx_lp_d < tx_ls + tx_ll:
if text_content[tx_lp_d] == '(':
break
tx_lp_d += 1
if tx_lp_d < tx_ls + tx_ll:
tx_sv_d: t.CChar = text_content[tx_lp_d + 1]
text_content[tx_lp_d + 1] = '\0'
_CommentOutDeclareInBuf(out_buf, text_content + tx_at_d)
text_content[tx_lp_d + 1] = tx_sv_d
# 处理:@global = ...
# stub 中可能是 external 声明或完整定义,以 text 中的定义为准
# 找到 stub 中同名全局变量行并注释掉,然后追加 text 中的
if tx_ll > 0 and text_content[tx_ls] == '@':
tx_eq: t.CSizeT = tx_ls + 1
while tx_eq < tx_ls + tx_ll:
if text_content[tx_eq] == '=':
break
tx_eq += 1
if tx_eq < tx_ls + tx_ll:
tx_sv2: t.CChar = text_content[tx_eq + 1]
text_content[tx_eq + 1] = '\0'
_CommentOutGlobalInBuf(out_buf, text_content + tx_ls)
text_content[tx_eq + 1] = tx_sv2
# 去重:%type = ...
if tx_ll > 0 and text_content[tx_ls] == '%':
tx_eq2: t.CSizeT = tx_ls + 1
while tx_eq2 < tx_ls + tx_ll:
if text_content[tx_eq2] == '=':
break
tx_eq2 += 1
if tx_eq2 < tx_ls + tx_ll:
tx_sv3: t.CChar = text_content[tx_eq2 + 1]
text_content[tx_eq2 + 1] = '\0'
if string.strstr(out_buf, text_content + tx_ls) is not None:
text_content[tx_eq2 + 1] = tx_sv3
continue
text_content[tx_eq2 + 1] = tx_sv3
# 追加
if out_pos + tx_ll + 2 < out_size:
string.strncpy(out_buf + out_pos, text_content + tx_ls, tx_ll)
out_pos += tx_ll
out_buf[out_pos] = '\n'
out_pos += 1
out_buf[out_pos] = '\0'
stdlib.free(text_content)
if deps_buf is not None:
stdlib.free(deps_buf)
if sha1_arr is not None:
stdlib.free(sha1_arr)
if mod_arr is not None:
stdlib.free(mod_arr)
return out_pos