TermiNexus
3e1e1fc1b3
match：case 数超过 MAX_MATCH_CASES 时 fail-fast（原先 break 会静默少分支，need_case 检查也发现不了）。vtable：初始化串按真实方法表动态分配，_vtable_strcat 溢出改 fatal；VTBL_METHOD_MAX 超限不再 break（原先子类 override 静默失效）。符号名：coro_pool_/__tls_init_done_/__decor_rec_/闭包 fn_ptr 的定义与引用改为按实际长度分配，消除 96 vs 128、64 vs 128 的不对称。Stub：_PyiEmit 溢出改 VLogger.error（半截 .vpi 会被当完整文件解析）；VpiReader 参数段/返回注解缓存槽放不下时 fail-fast，不再丢段导致形参注解错位。Includes：文件超过 1MiB 拒绝哈希（SHA1 冲突会让两个模块共用缓存）；add_file_entry 失败不再被忽略。f-string：格式串按字面量/转换符实际长度分配，字面量与 format_spec 溢出 fail-fast。
694 lines
38 KiB
Plaintext
694 lines
38 KiB
Plaintext
import lib.core.Gen.Generator as Generator
|
|
import t, c
|
|
from stdint import *
|
|
import ast
|
|
import memhub
|
|
import hashtable
|
|
import string
|
|
import stdlib
|
|
import viperlib
|
|
import stdio
|
|
import sys
|
|
import win32.fileio as fileio
|
|
import lib.Stub.StubMerger as StubMerger
|
|
import lib.core.CompilerFactory as CompilerFactory
|
|
import lib.core.Gen.Translator as HT
|
|
import lib.Stub.StubMerger.StubMergerStore as StubMergerStore
|
|
import lib.core.VLogger as VLogger
|
|
|
|
|
|
# ImportsGenMixin - import handling mixin. Maintains the imported_modules/from_imports
|
|
# tables, re-export registration, relative import resolution and module SHA1 checks.
|
|
|
|
|
|
# Constants (kept consistent with HandlesImports)
|
|
define MAX_REEXPORT = 512
|
|
|
|
|
|
# ImportsGenMixin - import handling
|
|
novirtual class ImportsGenMixin:
|
|
"""Import handling mixin."""
|
|
|
|
# --- Global re-export map tables (_g_reexport_sha1s/funcs/srcs/count) ---
|
|
Gen: Generator.Generator*
|
|
ReexportSha1s: bytes
|
|
ReexportFuncs: bytes
|
|
ReexportSrcs: bytes
|
|
ReexportCount: int
|
|
|
|
# --- Import lookup index (turns O(string) scanning into O(name) hashing) ---
|
|
# Bound to the current string pointer; a mismatch falls back to scanning. PermPool-allocated.
|
|
CurImportedModulesPtr: str
|
|
CurImportedModulesNames: hashtable.HashTable*
|
|
CurFromImportsPtr: str
|
|
CurFromImportsMods: hashtable.HashTable*
|
|
CurFromImportsOrigs: hashtable.HashTable*
|
|
|
|
# Note: CurrentFile is not declared here (TypeGenMixin owns it; a duplicate would
|
|
# misalign fields). Access goes through HandlesType to avoid a circular import.
|
|
|
|
# _index_imported_module - incrementally maintain the imported_modules index.
|
|
# Reuse the bound index when old_ptr matches, else unbind and fall back to scanning.
|
|
def _index_imported_module(self, old_ptr: str, new_ptr: str, name_ptr: str) -> None:
|
|
"""Incrementally update the imported_modules index, reusing it when possible."""
|
|
if new_ptr is None or name_ptr is None: return
|
|
perm: memhub.MemBuddy* = CompilerFactory.factory.PermPool
|
|
if perm is None: return
|
|
idx: hashtable.HashTable* = None
|
|
if old_ptr is None:
|
|
# New file / chain head: build a fresh index (the old one is discarded with its string)
|
|
idx = hashtable.HashTable(perm)
|
|
elif old_ptr == self.CurImportedModulesPtr and self.CurImportedModulesNames is not None:
|
|
# Normal path: reuse the current string's index and append (no rebuild, O(imports) memory)
|
|
idx = self.CurImportedModulesNames
|
|
else:
|
|
# Old string is not the bound one (unexpected source): unbind and fall back to scanning
|
|
self.CurImportedModulesPtr = None
|
|
self.CurImportedModulesNames = None
|
|
return
|
|
if idx is None:
|
|
self.CurImportedModulesPtr = None
|
|
self.CurImportedModulesNames = None
|
|
return
|
|
# Skip if the name already exists (duplicate import); otherwise write (key/value point inside new_ptr)
|
|
if idx.__contains__(name_ptr) == 0: idx.set_str(name_ptr, name_ptr)
|
|
self.CurImportedModulesPtr = new_ptr
|
|
self.CurImportedModulesNames = idx
|
|
|
|
# _index_from_import - incrementally maintain the from_imports index. entry_off is
|
|
# the new entry's offset; module/orig tables key by local_name (first hit wins).
|
|
def _index_from_import(self, old_ptr: str, new_ptr: str, entry_off: sizet, local_name: str, module_name: str, original_name: str) -> None:
|
|
"""Incrementally update the from_imports index, reusing it when possible."""
|
|
if new_ptr is None or local_name is None or module_name is None: return
|
|
perm: memhub.MemBuddy* = CompilerFactory.factory.PermPool
|
|
if perm is None: return
|
|
idx_m: hashtable.HashTable* = None
|
|
idx_o: hashtable.HashTable* = None
|
|
if old_ptr is None:
|
|
idx_m = hashtable.HashTable(perm)
|
|
idx_o = hashtable.HashTable(perm)
|
|
elif old_ptr == self.CurFromImportsPtr and self.CurFromImportsMods is not None and self.CurFromImportsOrigs is not None:
|
|
idx_m = self.CurFromImportsMods
|
|
idx_o = self.CurFromImportsOrigs
|
|
else:
|
|
self.CurFromImportsPtr = None
|
|
self.CurFromImportsMods = None
|
|
self.CurFromImportsOrigs = None
|
|
return
|
|
if idx_m is None or idx_o is None:
|
|
self.CurFromImportsPtr = None
|
|
self.CurFromImportsMods = None
|
|
self.CurFromImportsOrigs = None
|
|
return
|
|
local_len: sizet = string.strlen(local_name)
|
|
mod_len: sizet = string.strlen(module_name)
|
|
# The hash key must be NUL-terminated: the flat string's local segment ends with ':',
|
|
# so hashing it directly never matches. Allocate a stable NUL-terminated copy.
|
|
key_ptr: str = perm.alloc(local_len + 1)
|
|
if key_ptr is None:
|
|
self.CurFromImportsPtr = None
|
|
self.CurFromImportsMods = None
|
|
self.CurFromImportsOrigs = None
|
|
return
|
|
string.strncpy(key_ptr, local_name, local_len)
|
|
key_ptr[local_len] = '\0'
|
|
mod_ptr: str = new_ptr + entry_off + local_len + 1
|
|
if local_len == 1 and local_name[0] == '*':
|
|
# star import: the last entry's module name wins (overwrite, matching star_mod)
|
|
idx_m.set_str(key_ptr, mod_ptr)
|
|
else:
|
|
if idx_m.__contains__(key_ptr) == 0: idx_m.set_str(key_ptr, mod_ptr)
|
|
if idx_o.__contains__(key_ptr) == 0:
|
|
if original_name is not None and string.strcmp(original_name, local_name) != 0 and string.strlen(original_name) > 0: idx_o.set_str(key_ptr, mod_ptr + mod_len + 1)
|
|
else:
|
|
# First occurrence has no alias (or empty alias) -> query returns None, matching scanning
|
|
idx_o.set_str(key_ptr, None)
|
|
self.CurFromImportsPtr = new_ptr
|
|
self.CurFromImportsMods = idx_m
|
|
self.CurFromImportsOrigs = idx_o
|
|
|
|
# Add an imported module name
|
|
def add_imported_module(self, pool: memhub.MemBuddy*, imported_modules: str, name: str) -> str:
|
|
"""Record an imported module name and return the new imported_modules string."""
|
|
if name is None: return imported_modules
|
|
old_len: sizet = 0
|
|
if imported_modules is not None: old_len = string.strlen(imported_modules)
|
|
buf: str = StubMergerStore.AppendImportedModule(pool, imported_modules, name)
|
|
if buf is None: return imported_modules
|
|
if imported_modules is None:
|
|
# Index: first module name (key points inside buf)
|
|
self._index_imported_module(None, buf, buf)
|
|
else:
|
|
# Index: reuse the old string's index and append the new name (key points inside buf)
|
|
self._index_imported_module(imported_modules, buf, buf + old_len + 1)
|
|
return buf
|
|
|
|
# Check whether a module has been imported
|
|
def is_module_imported(self, imported_modules: str, name: str) -> int:
|
|
"""Check whether a module is imported, matching on word boundaries."""
|
|
if name is None or imported_modules is None: return 0
|
|
# [Index fast path] When the input is the bound current string, do an O(name) hash
|
|
# lookup; otherwise fall back to the scanning below, semantics unchanged.
|
|
if imported_modules == self.CurImportedModulesPtr and self.CurImportedModulesNames is not None:
|
|
if self.CurImportedModulesNames.__contains__(name) != 0: return 1
|
|
# Index miss never returns 0 (an incomplete index would misreport "not imported");
|
|
# fall through to scanning on the real string content (correctness first).
|
|
name_len: sizet = string.strlen(name)
|
|
cur: str = imported_modules
|
|
total_len: sizet = string.strlen(imported_modules)
|
|
ci: sizet = 0
|
|
while ci < total_len:
|
|
while ci < total_len and cur[ci] == ' ': ci += 1
|
|
if ci >= total_len: break
|
|
word_start: sizet = ci
|
|
while ci < total_len and cur[ci] != ' ': ci += 1
|
|
word_len: sizet = ci - word_start
|
|
if word_len == name_len:
|
|
Matched: int = 1
|
|
for ei in range(int(name_len)):
|
|
if cur[word_start + ei] != name[ei]:
|
|
Matched = 0
|
|
break
|
|
if Matched == 1: return 1
|
|
return 0
|
|
|
|
# Check whether a module is imported or globally registered (package submodule resolution)
|
|
def is_module_imported_or_known(self, imported_modules: str, name: str) -> int:
|
|
"""Return 1 when the module is imported or reachable through an imported package."""
|
|
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: sizet = string.strlen(name)
|
|
for dot_pos in range(int(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
|
|
return 0
|
|
|
|
# Add a from-import name mapping
|
|
def add_from_import(self, pool: memhub.MemBuddy*, from_imports: str, local_name: str, module_name: str, original_name: str = None) -> str:
|
|
"""Add a from-import mapping and return the new from_imports string."""
|
|
if local_name is None or module_name is None: return from_imports
|
|
old_len: sizet = 0
|
|
if from_imports is not None: old_len = string.strlen(from_imports)
|
|
buf: str = StubMergerStore.AppendFromImport(pool, from_imports, local_name, module_name, original_name)
|
|
if buf is None: return from_imports
|
|
# Index: the first entry has entry_off=0; an appended entry starts at buf[old_len+1]
|
|
entry_off: sizet = 0
|
|
if from_imports is not None: entry_off = old_len + 1
|
|
self._index_from_import(from_imports, buf, entry_off, local_name, module_name, original_name)
|
|
return buf
|
|
|
|
# Look up a from-import name and return the original function name (alias case):
|
|
# for `from X import Y as Z` return Y; return None when there is no alias.
|
|
def lookup_from_import_original(self, from_imports: str, local_name: str) -> str:
|
|
"""Return the original function name for a from-import alias, or None."""
|
|
if local_name is None or from_imports is None: return None
|
|
# [Index fast path] When the input is the bound current string, do an O(name) hash
|
|
# lookup; else fall back to scanning. orig None means a first occurrence with no alias.
|
|
if from_imports == self.CurFromImportsPtr and self.CurFromImportsOrigs is not None:
|
|
if self.CurFromImportsOrigs.__contains__(local_name) != 0:
|
|
v_o: t.CPtr = self.CurFromImportsOrigs.__getitem__(local_name)
|
|
if v_o is not None: return rawptr(v_o)
|
|
# Index miss or None value never returns None directly (the real string may still
|
|
# contain the name); fall through to scanning (correctness first).
|
|
name_len: sizet = string.strlen(local_name)
|
|
cur: str = from_imports
|
|
ci: sizet = 0
|
|
total_len: sizet = string.strlen(from_imports)
|
|
while ci < total_len:
|
|
# Skip leading spaces
|
|
while ci < total_len and cur[ci] == ' ': ci += 1
|
|
if ci >= total_len: break
|
|
# Find the first ':'
|
|
colon1: sizet = ci
|
|
while colon1 < total_len and cur[colon1] != ':' and cur[colon1] != ' ': colon1 += 1
|
|
if colon1 >= total_len or cur[colon1] != ':': break
|
|
entry_name_len: sizet = colon1 - ci
|
|
# Skip star imports
|
|
if entry_name_len == 1 and cur[ci] == '*':
|
|
ci = colon1
|
|
while ci < total_len and cur[ci] != ' ': ci += 1
|
|
continue
|
|
# Compare the name
|
|
if entry_name_len == name_len:
|
|
Matched: int = 1
|
|
for ei in range(int(name_len)):
|
|
if cur[ci + ei] != local_name[ei]:
|
|
Matched = 0
|
|
break
|
|
if Matched == 1:
|
|
# Match found; check for a third field (the original name)
|
|
pos: sizet = colon1 + 1
|
|
# Skip the module name
|
|
while pos < total_len and cur[pos] != ':' and cur[pos] != ' ' and cur[pos] != '\0': pos += 1
|
|
if pos < total_len and cur[pos] == ':':
|
|
# Third field present: original_name
|
|
orig_start: sizet = pos + 1
|
|
orig_end: sizet = orig_start
|
|
while orig_end < total_len and cur[orig_end] != ' ' and cur[orig_end] != '\0': orig_end += 1
|
|
orig_len: sizet = orig_end - orig_start
|
|
if orig_len > 0:
|
|
orig_buf: str = cur + orig_start
|
|
# Return an interior pointer (caller keeps from_imports alive during use)
|
|
return orig_buf
|
|
return None
|
|
return None
|
|
# Advance to the next entry
|
|
ci = colon1
|
|
while ci < total_len and cur[ci] != ' ': ci += 1
|
|
return None
|
|
|
|
# Look up a from-import name and return the module name, or None.
|
|
# allow_star_fallback=1 permits a star-import fallback; pass 0 to disable it.
|
|
def lookup_from_import(self, from_imports: str, name: str, allow_star_fallback: int = 1) -> str:
|
|
"""Return the module name for a from-import, with optional star fallback."""
|
|
if name is None or from_imports is None: return None
|
|
# [Index fast path] When the input is the bound current string, do an O(name) hash
|
|
# lookup; else fall back to scanning. name=="*" skips exact match and goes to star.
|
|
if from_imports == self.CurFromImportsPtr and self.CurFromImportsMods is not None:
|
|
name_is_star: int = 0
|
|
if string.strlen(name) == 1 and name[0] == '*': name_is_star = 1
|
|
if name_is_star == 0:
|
|
v_f: t.CPtr = self.CurFromImportsMods.__getitem__(name)
|
|
if v_f is not None: return rawptr(v_f)
|
|
if allow_star_fallback != 0:
|
|
sv_f: t.CPtr = self.CurFromImportsMods.__getitem__("*")
|
|
if sv_f is not None: return rawptr(sv_f)
|
|
# Neither exact nor star hit in the index never returns None (an incomplete index
|
|
# misses imported names, causing _op_head NameError); fall through to scanning.
|
|
name_len: sizet = string.strlen(name)
|
|
cur: str = from_imports
|
|
ci: sizet = 0
|
|
total_len: sizet = string.strlen(from_imports)
|
|
star_mod: str = None
|
|
while ci < total_len:
|
|
# Skip leading spaces
|
|
while ci < total_len and cur[ci] == ' ': ci += 1
|
|
if ci >= total_len: break
|
|
# Find the ':' position
|
|
colon_pos: sizet = 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: sizet = colon_pos - ci
|
|
# Detect star import ("*:module")
|
|
if entry_name_len == 1 and cur[ci] == '*':
|
|
mod_start: sizet = colon_pos + 1
|
|
mod_end: sizet = 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
|
|
# Compare the name
|
|
elif entry_name_len == name_len:
|
|
Matched: int = 1
|
|
for ei in range(int(name_len)):
|
|
if cur[ci + ei] != name[ei]:
|
|
Matched = 0
|
|
break
|
|
if Matched == 1:
|
|
mod_start2: sizet = colon_pos + 1
|
|
mod_end2: sizet = mod_start2
|
|
# The module name ends at a space, null, or a second ':' (alias separator)
|
|
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
|
|
# Advance to the next entry
|
|
ci = colon_pos
|
|
while ci < total_len and cur[ci] != ' ': ci += 1
|
|
# No exact match; fall back to star import only when allowed
|
|
if allow_star_fallback == 0: return None
|
|
return star_mod
|
|
|
|
# register_reexport - register a re-export mapping (called from HandleImportFromNames
|
|
# for `from .X import Y`). Idempotent; memory is stdlib.malloc'd to persist across phases.
|
|
def register_reexport(self, mod_sha1: str, func_name: str, source_module: str) -> int:
|
|
"""Register a re-export mapping; returns 0 on success, -1 on failure."""
|
|
if mod_sha1 is None or func_name is None or source_module is None: return -1
|
|
# fail-fast: the table is a flat slab with fixed slots (17 bytes for the SHA1, 64 for the
|
|
# name/module). strncpy used to clamp silently, after which lookup_reexport's full-name
|
|
# strcmp never matched (or matched another module sharing the 63-byte prefix).
|
|
if string.strlen(mod_sha1) > 16:
|
|
self.Gen.TypeGenErr.fatal_error(None, "re-export module SHA1 is longer than its 16-byte table slot", "OverflowError")
|
|
if string.strlen(func_name) > 62:
|
|
self.Gen.TypeGenErr.fatal_error(None, f"re-exported name '{func_name}' exceeds the 62-byte table slot; shorten the identifier", "OverflowError")
|
|
if string.strlen(source_module) > 62:
|
|
self.Gen.TypeGenErr.fatal_error(None, f"re-export source module '{source_module}' exceeds the 62-byte table slot; shorten the module name", "OverflowError")
|
|
# Lazy initialization
|
|
if self.ReexportSha1s is None:
|
|
self.ReexportSha1s = CompilerFactory.factory.PermPool.alloc(MAX_REEXPORT * 17)
|
|
self.ReexportFuncs = CompilerFactory.factory.PermPool.alloc(MAX_REEXPORT * 64)
|
|
self.ReexportSrcs = CompilerFactory.factory.PermPool.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
|
|
# Idempotency check: look for an existing identical mapping
|
|
for i in range(int(self.ReexportCount)):
|
|
sidx: sizet = sizet(i) * 17
|
|
fidx: sizet = sizet(i) * 64
|
|
if string.strcmp(self.ReexportSha1s + sidx, mod_sha1) == 0:
|
|
if string.strcmp(self.ReexportFuncs + fidx, func_name) == 0:
|
|
# Already exists; update source_module (64-byte slot, strncpy-clamped)
|
|
string.strncpy(self.ReexportSrcs + fidx, source_module, 63)
|
|
self.ReexportSrcs[fidx + 63] = '\0'
|
|
return 0
|
|
# Add a new entry (fixed 17/64-byte slots; unbounded strcpy would overrun, so all strncpy)
|
|
idx2: sizet = sizet(self.ReexportCount) * 17
|
|
fidx2: sizet = sizet(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 - given a module SHA1 and function name, return the source module
|
|
# name (e.g. "ast.lexer") or None; the caller resolves its SHA1 separately.
|
|
def lookup_reexport(self, mod_sha1: str, func_name: str) -> str:
|
|
"""Return the source module name for a re-export mapping, or 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(int(self.ReexportCount)):
|
|
sidx: sizet = sizet(i) * 17
|
|
fidx: sizet = sizet(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 - resolve a relative import into a full module name:
|
|
# `from .X import Y` -> "pkg.X", `from . import Y` -> "pkg", `from ..X` -> "pkg.X".
|
|
def _resolve_relative_module(self, pool: memhub.MemBuddy*, current_package: str, level: int, module: str) -> str:
|
|
"""Resolve a relative import into a full module name."""
|
|
if level <= 0: return module
|
|
if current_package is None:
|
|
# No package context (top-level includes/App files): `from .X import Y` means
|
|
# "the X module in the same/root directory", so returning module is correct.
|
|
return module
|
|
|
|
# Climb (level-1) levels up from current_package
|
|
pkg: str = current_package
|
|
up: int = level - 1
|
|
while up > 0:
|
|
pkg_len: sizet = string.strlen(pkg)
|
|
last_dot: sizet = 0
|
|
found: int = 0
|
|
for i in range(int(pkg_len)):
|
|
if pkg[i] == '.':
|
|
last_dot = i
|
|
found = 1
|
|
if found == 0:
|
|
# No more parents (the relative level exceeds the package depth): fall back to
|
|
# the original module; a later import SHA1 check reports "module not found".
|
|
pkg = None
|
|
break
|
|
# Truncate at the last '.'
|
|
new_pkg: str = pool.alloc(last_dot + 1)
|
|
if new_pkg is None:
|
|
self.Gen.TypeGenErr.fatal_error(None, f"relative import resolution failed: 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 -> the module is the package itself
|
|
return pkg
|
|
if pkg is None:
|
|
# The package climbed to empty (relative import out of range): fall back to module
|
|
return module
|
|
# Concatenate pkg + "." + module
|
|
pkg_len2: sizet = string.strlen(pkg)
|
|
mod_len: sizet = string.strlen(module)
|
|
buf: str = pool.alloc(pkg_len2 + 1 + mod_len + 1)
|
|
if buf is None:
|
|
self.Gen.TypeGenErr.fatal_error(None, f"relative import resolution failed: 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 - derive the package name from a relative path by taking
|
|
# the directory and replacing / and \ with '.'. Top-level files (no separator) -> None.
|
|
def compute_package_from_relpath(self, pool: memhub.MemBuddy*, rel_path: str) -> str:
|
|
"""Derive the package name from a relative path by replacing separators with dots."""
|
|
if rel_path is None: return None
|
|
rlen: sizet = string.strlen(rel_path)
|
|
# Find the last / or \
|
|
last_sep: sizet = 0
|
|
found: int = 0
|
|
for i in range(int(rlen)):
|
|
ch: char = rel_path[i]
|
|
if ch == '/' or ch == '\\':
|
|
last_sep = i
|
|
found = 1
|
|
if found == 0:
|
|
# No directory separator -> top-level file, no package
|
|
return None
|
|
# Copy the directory part, replacing / and \ with .
|
|
dir_len: sizet = last_sep
|
|
buf: str = pool.alloc(dir_len + 1)
|
|
if buf is None: return None
|
|
for j in range(int(dir_len)):
|
|
ch2: char = rel_path[j]
|
|
if ch2 == '/' or ch2 == '\\': buf[j] = '.'
|
|
else: buf[j] = ch2
|
|
buf[dir_len] = '\0'
|
|
return buf
|
|
|
|
# set_current_file - set the file being compiled, matching TypeGenMixin's behavior;
|
|
# delegated through HandlesType (lazy import avoids a circular dependency).
|
|
def set_current_file(self, file_path: str) -> None:
|
|
"""Set the file currently being compiled, synced with HandlesType."""
|
|
# File-level invalidation: the index binds a string pointer from the previous file's
|
|
# arena that may be reused after reset, so unbind; add_* rebuilds it for this file.
|
|
self.CurImportedModulesPtr = None
|
|
self.CurImportedModulesNames = None
|
|
self.CurFromImportsPtr = None
|
|
self.CurFromImportsMods = None
|
|
self.CurFromImportsOrigs = None
|
|
self.Gen.TypeGenResolve.set_current_file(file_path)
|
|
|
|
# _report_import_error - local Python-style error reporter to avoid a circular import
|
|
# with HandlesType. Emits a Traceback/File/ImportError block and aborts.
|
|
def _report_import_error(self, node: ast.AST*, msg: str) -> None:
|
|
"""Report a Python-style import error and abort."""
|
|
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)
|
|
# Show the source line's content (read line lineno for quick location)
|
|
if lineno > 0 and file_name is not None:
|
|
f_err: fileio.File* = 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: sizet = 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)
|
|
# Termination convergence: longjmp back to the REPL loop in a session, else abort
|
|
VLogger.repl_abort()
|
|
|
|
# _is_local_module_file - a sibling {mod_name}.vp/.py exists next to the current file
|
|
# (e.g. caller1.vp importing g4node); such modules are exempt from the SHA1 check.
|
|
def _is_local_module_file(self, mod_name: str) -> int:
|
|
"""Return 1 when a sibling {mod_name}.vp/.py exists next to the current file."""
|
|
cur: str = self.Gen.TypeGenResolve.get_current_file()
|
|
if cur is None or mod_name is None: return 0
|
|
cur_len: sizet = string.strlen(cur)
|
|
if cur_len == 0: return 0
|
|
sep_pos: sizet = cur_len
|
|
while sep_pos > 0:
|
|
sep_pos -= 1
|
|
ch: char = cur[sep_pos]
|
|
if ch == '/' or ch == '\\':
|
|
sep_pos += 1
|
|
break
|
|
mod_len: sizet = string.strlen(mod_name)
|
|
path_len: sizet = sep_pos + mod_len + 5
|
|
path_buf: bytes = stdlib.malloc(path_len + 1)
|
|
if path_buf is None: return 0
|
|
# .vp file (Viper sources are .vp only, no .py fallback)
|
|
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* = 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 - validate at import time that the module SHA1 is registered
|
|
# (fast-fail). Reports immediately instead of misleading the user later.
|
|
def _check_import_sha1(self, node: ast.AST*, mod_name: str) -> int:
|
|
"""Fail-fast when an imported module's SHA1 is not registered."""
|
|
if mod_name is None: return 1
|
|
# t/c are internal libraries exempt from includes scanning; so are ctraits (a .vpi
|
|
# compile-time macro lib) and viperbuild (a compiler builtin macro lib).
|
|
if string.strcmp(mod_name, "t") == 0 or string.strcmp(mod_name, "c") == 0 or string.strcmp(mod_name, "ctraits") == 0: return 1
|
|
if string.strcmp(mod_name, "viperbuild") == 0: return 1
|
|
mod_sha1: str = StubMergerStore.LookupSha1ByModName(mod_name)
|
|
if mod_sha1 is None:
|
|
# App sibling exemption: a {mod_name}.vp/.py exists next to the current file
|
|
# (SHA1 lives in the project file list, not the includes store).
|
|
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, "module '%s' has no registered SHA1 (the includes scan did not find this module; check the includes path configuration in project.json)", mod_name)
|
|
self._report_import_error(node, err_buf)
|
|
else: self._report_import_error(node, "imported module SHA1 not registered (the includes scan did not find this module)")
|
|
return 1
|
|
|
|
# translate_import - handle import statements and update trans._imported_modules
|
|
def translate_import(self, trans: HT.Translator*, node: ast.AST*) -> int:
|
|
"""Translate an import statement and update trans._imported_modules."""
|
|
imp: ast.Import* = ast.Import*(node)
|
|
if imp is None: return 0
|
|
names: list[ast.AST*]* = imp.names
|
|
if names is None: return 0
|
|
nn: sizet = len(names)
|
|
pool_val: memhub.MemBuddy* = trans.Pool
|
|
ctx: StubMergerStore.ImportCtx* = StubMergerStore.AllocImportCtx(pool_val, trans._imported_modules, trans._from_imports)
|
|
if ctx is None: return 0
|
|
for ni in range(int(nn)):
|
|
alias: ast.Alias* = ast.Alias*(names.get(ni))
|
|
if alias is not None and alias.name is not None:
|
|
# Validate the module SHA1 at import time (fast-fail, full translation only):
|
|
# declare_only=1/2 is the pre-registration scan and would false-positive.
|
|
if trans._declare_only == 0: self._check_import_sha1(node, alias.name)
|
|
StubMergerStore.NotePlainImport(pool_val, ctx, alias.name, alias.asname)
|
|
trans._imported_modules = ctx.ImportedModules
|
|
trans._from_imports = ctx.FromImports
|
|
return 0
|
|
|
|
# translate_import_from_module - handle the module part of `from X import Y`.
|
|
# Relative imports (level>0) resolve via trans.CurrentPackage to a full module name.
|
|
def translate_import_from_module(self, trans: HT.Translator*, node: ast.AST*) -> int:
|
|
"""Translate the module part of a from-import statement."""
|
|
impf: ast.ImportFrom* = ast.ImportFrom*(node)
|
|
if impf is None: return 0
|
|
# Resolve the module name: relative imports need the package prefix
|
|
resolved: str = self._resolve_relative_module(trans.Pool, trans.CurrentPackage, impf.level, impf.module)
|
|
if resolved is not None:
|
|
# Validate the module SHA1 at import time (fast-fail, full translation only). For
|
|
# `from . import Y` resolved is a package directory, not a file, so skip the check.
|
|
if trans._declare_only == 0 and impf.module is not None: self._check_import_sha1(node, resolved)
|
|
trans._imported_modules = self.add_imported_module(trans.Pool, trans._imported_modules, resolved)
|
|
# For `from . import Y` resolved is the package name; each name is a submodule.
|
|
# Add the full submodule names or cross-module resolution of syscall.* fails.
|
|
if impf.module is None:
|
|
names_m: list[ast.AST*]* = impf.names
|
|
if names_m is not None:
|
|
nn_m: sizet = len(names_m)
|
|
for ni_m in range(int(nn_m)):
|
|
alias_m: ast.Alias* = ast.Alias*(names_m.get(ni_m))
|
|
if alias_m is not None and alias_m.name is not None:
|
|
rl_m: sizet = string.strlen(resolved)
|
|
rn_m: sizet = 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)
|
|
# Y may be a package __init__.vp symbol (class/constant) with no
|
|
# file or SHA1; validate/register only for real submodules or local files.
|
|
if (StubMergerStore.LookupSha1ByModName(jb_m) is not None or self._is_local_module_file(alias_m.name) != 0):
|
|
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 handles the name part of `from X import Y` below.
|
|
# _CheckStaticFromImport fail-fasts on `from mod import name` hitting a static symbol.
|
|
def _CheckStaticFromImport(self, trans: HT.Translator*, fi_target: str, orig_name: str, node: ast.AST*) -> int:
|
|
if trans is None or fi_target is None or orig_name is None: return 0
|
|
if trans.ModuleSha1 is None: return 0
|
|
# Fast short-circuit: zero overhead when the project has no static symbols
|
|
if self.Gen.CallGenBase.StaticCount <= 0: return 0
|
|
sf_sha1: str = self.Gen.StructGen._resolve_module_sha1(fi_target, trans._from_imports)
|
|
if sf_sha1 is None: return 0
|
|
if string.strncmp(sf_sha1, trans.ModuleSha1, 16) == 0: return 0
|
|
if self.Gen.CallGenBase.IsStaticSymbol(sf_sha1, orig_name) == 0: return 0
|
|
self.Gen.TypeGenErr.fatal_error(node, f"'{fi_target}.{orig_name}' is a static symbol; cross-module use is forbidden (static is visible only within the defining module)", "NameError")
|
|
return 0
|
|
|
|
|
|
def translate_import_from_names(self, trans: HT.Translator*, node: ast.AST*) -> int:
|
|
"""Translate the name part of a from-import statement."""
|
|
impf: ast.ImportFrom* = ast.ImportFrom*(node)
|
|
if impf is None: return 0
|
|
# Resolve the module name: relative imports need the package prefix
|
|
resolved: str = self._resolve_relative_module(trans.Pool, trans.CurrentPackage, impf.level, impf.module)
|
|
if resolved is not None:
|
|
names: list[ast.AST*]* = impf.names
|
|
if names is not None:
|
|
nn: sizet = len(names)
|
|
for ni in range(int(nn)):
|
|
alias: ast.Alias* = ast.Alias*(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
|
|
# For `from . import Y`, _resolve_relative_module returns the package name
|
|
# but Y is a submodule, so map local_name -> the full submodule name.
|
|
fi_target: str = resolved
|
|
if impf.module is None and resolved is not None:
|
|
fl_t: sizet = string.strlen(resolved)
|
|
fn_t: sizet = 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)
|
|
# Static symbols (forbidden cross-module): `from mod import name` hitting one
|
|
# fail-fasts. Logic stays in _CheckStaticFromImport (inlining breaks inference).
|
|
self._CheckStaticFromImport(trans, fi_target, alias.name, node)
|
|
# Namespace isolation: mark the from-imported name as a visible struct
|
|
self.Gen.StructGen.add_visible_struct(trans.Pool, local_name)
|
|
# Register the re-export mapping (current module re-exports local_name from
|
|
# fi_target) so cross-module module.func() resolves to the right SHA1.
|
|
if trans.ModuleSha1 is not None: self.register_reexport(trans.ModuleSha1, local_name, fi_target)
|
|
return 0
|
|
|
|
def InitImportsGen(self) -> int:
|
|
"""Initialize the ImportsGen fields, called by Generator at construction."""
|
|
if self is None: return 1
|
|
self.ReexportSha1s = None
|
|
self.ReexportFuncs = None
|
|
self.ReexportSrcs = None
|
|
self.ReexportCount = 0
|
|
# Import lookup index fields (created incrementally by add_*, PermPool-allocated)
|
|
self.CurImportedModulesPtr = None
|
|
self.CurImportedModulesNames = None
|
|
self.CurFromImportsPtr = None
|
|
self.CurFromImportsMods = None
|
|
self.CurFromImportsOrigs = None
|
|
return 0
|