refactor(includes): 适配新指针口径与语法,rawptr[T] → T* / t.CPtr → rawptr

- json/__init__.vp、math.vp:rawptr[T] 改为 T*
- ast/base.vp、ast/stmts.vp:@t.CVTable 改为 virtual class
- argparse.vp:t.CPtr(...) 改为 str*(...)
- stdarg.vp:arg(type: t.CType) 改为 t.CPtr
- stdint.vp:FSIZE_t/LBA_t 换成 LPCSTR
- memhub.vp:import 整理、去掉 t.CDefine 前缀、t.CPtr 改 rawptr、格式对齐
- ast/parser.vp、math.vp、linkedlist.vp:对齐与格式清理
This commit is contained in:
TermiNexus committed 2026-10-01 18:05:05 +08:00
1 parent d8c02cff73
commit 14221b9414
8 files changed
+184 -219

No files matched your search

+4 -4
View File
@@ -294,7 +294,7 @@ class ArgumentParser:
entry: str | t.CPtr = None
while idx < argc:
entry = t.CPtr(t.CUInt64T(argv) + idx * 8)
entry = str*(t.CUInt64T(argv) + idx * 8)
token = entry[0]
if token is None:
break
@@ -303,7 +303,7 @@ class ArgumentParser:
if token[0] == '-' and token[1] == '-' and token[2] == 0:
idx += 1
while idx < argc:
entry = t.CPtr(t.CUInt64T(argv) + idx * 8)
entry = str*(t.CUInt64T(argv) + idx * 8)
token = entry[0]
found = self._find_positional(pos_idx)
if found < 0:
@@ -351,7 +351,7 @@ class ArgumentParser:
stdio.printf("error: %s requires a value\n", arg.name)
release(args)
return None
entry = t.CPtr(t.CUInt64T(argv) + (idx + 1) * 8)
entry = str*(t.CUInt64T(argv) + (idx + 1) * 8)
value = entry[0]
self._store_value(args, arg, value)
idx += 2
@@ -372,7 +372,7 @@ class ArgumentParser:
stdio.printf("error: %s requires a value\n", arg.name)
release(args)
return None
entry = t.CPtr(t.CUInt64T(argv) + (idx + 1) * 8)
entry = str*(t.CUInt64T(argv) + (idx + 1) * 8)
value = entry[0]
self._store_value(args, arg, value)
idx += 2
+3 -6
View File
@@ -310,8 +310,7 @@ def _dump_op_list(lst: t.CPtr, buf: str, size: sizet, pos: sizet) -> sizet:
# Child nodes are stored in a list[AST | CPtr] container (O(1) random access, more efficient than linked lists)
# The parent field maintains a pointer to the parent (used for upward traversal)
# ============================================================
@t.CVTable
class AST:
virtual class AST:
"""AST node base class. All specific node classes inherit from this.
Fields:
@@ -368,8 +367,7 @@ class AST:
# (依赖 TPV isinstance 子类集合判定 + 三层 vtable 继承)。
# 不 override kind()(继承 AST 的 0),具体节点各自返回 ASTKind 值。
# ============================================================
@t.CVTable
class Stmt(AST):
virtual class Stmt(AST):
"""语句节点基类(Assign/If/For/While/Return/...)"""
@@ -382,8 +380,7 @@ class Stmt(AST):
# 命名说明:不能叫 Expr——ASTKind.Expr(base.vp L23)与 stmts.vp 的
# Expr 类已是"表达式语句"(Expression Statement)的既有语义。
# ============================================================
@t.CVTable
class ExprNode(AST):
virtual class ExprNode(AST):
"""表达式节点基类(Name/Constant/Call/BinOp/...)"""
+1 -3
View File
@@ -13,9 +13,7 @@ from .base import (
# ============================================================
# Module Node
# ============================================================
@t.CVTable
class Module(AST):
virtual class Module(AST):
"""Module(children: list[rawptr[AST]])"""
def __new__(self, pool: rawptr[memhub.MemManager]):
+22 -22
View File
@@ -33,7 +33,7 @@ class JsonValue:
def __new__(self, pool: rawptr[memhub.MemManager]):
"""从 mpool 分配 JsonValue 结构体内存并零初始化"""
ptr: rawptr[JsonValue] = pool.alloc(JsonValue.__sizeof__())
ptr: JsonValue* = pool.alloc(JsonValue.__sizeof__())
if ptr:
string.memset(ptr, 0, JsonValue.__sizeof__())
return ptr
@@ -83,7 +83,7 @@ class JsonValue:
return None
if key == None:
return None
cur: rawptr[JsonValue] = self.child
cur: JsonValue* = self.child
while cur != None:
if cur.key != None:
if string.strcmp(cur.key, key) == 0:
@@ -95,7 +95,7 @@ class JsonValue:
"""Python风格赋值:v["key"] = val"""
if self.vtype != JSON_OBJECT or key == None or val == None:
return
cur: rawptr[JsonValue] = self.child
cur: JsonValue* = self.child
while cur != None:
if cur.key != None:
if string.strcmp(cur.key, key) == 0:
@@ -120,7 +120,7 @@ class JsonValue:
"""按索引取数组元素:v.get_item(0)"""
if self.vtype != JSON_ARRAY and self.vtype != JSON_OBJECT:
return None
cur: rawptr[JsonValue] = self.child
cur: JsonValue* = self.child
i: sizet = 0
while cur != None:
if i == index:
@@ -137,7 +137,7 @@ class JsonValue:
if self.child == None:
self.child = child
else:
cur: rawptr[JsonValue] = self.child
cur: JsonValue* = self.child
while cur.next != None:
cur = cur.next
cur.next = child
@@ -148,35 +148,35 @@ class JsonValue:
# 构造函数:创建各类 JsonValue(从 mpool 分配)
# ============================================================
def null(pool: rawptr[memhub.MemManager]) -> rawptr[JsonValue]:
v: rawptr[JsonValue] = JsonValue(pool)
def null(pool: rawptr[memhub.MemManager]) -> JsonValue*:
v: JsonValue* = JsonValue(pool)
if v == None: return None
v.vtype = JSON_NULL
return v
def bool_val(pool: rawptr[memhub.MemManager], val: bool) -> rawptr[JsonValue]:
v: rawptr[JsonValue] = JsonValue(pool)
def bool_val(pool: rawptr[memhub.MemManager], val: bool) -> JsonValue*:
v: JsonValue* = JsonValue(pool)
if v == None: return None
v.vtype = JSON_BOOL
v.bool_val = 1 if val else 0
return v
def int_val(pool: rawptr[memhub.MemManager], val: int64) -> rawptr[JsonValue]:
v: rawptr[JsonValue] = JsonValue(pool)
def int_val(pool: rawptr[memhub.MemManager], val: int64) -> JsonValue*:
v: JsonValue* = JsonValue(pool)
if v == None: return None
v.vtype = JSON_INT
v.int_val = val
return v
def float_val(pool: rawptr[memhub.MemManager], val: double) -> rawptr[JsonValue]:
v: rawptr[JsonValue] = JsonValue(pool)
def float_val(pool: rawptr[memhub.MemManager], val: double) -> JsonValue*:
v: JsonValue* = JsonValue(pool)
if v == None: return None
v.vtype = JSON_FLOAT
v.float_val = val
return v
def string_val(pool: rawptr[memhub.MemManager], val: str) -> rawptr[JsonValue]:
v: rawptr[JsonValue] = JsonValue(pool)
def string_val(pool: rawptr[memhub.MemManager], val: str) -> JsonValue*:
v: JsonValue* = JsonValue(pool)
if v == None: return None
v.vtype = JSON_STRING
if val != None:
@@ -187,29 +187,29 @@ def string_val(pool: rawptr[memhub.MemManager], val: str) -> rawptr[JsonValue]:
v.str_val = buf
return v
def array(pool: rawptr[memhub.MemManager]) -> rawptr[JsonValue]:
v: rawptr[JsonValue] = JsonValue(pool)
def array(pool: rawptr[memhub.MemManager]) -> JsonValue*:
v: JsonValue* = JsonValue(pool)
if v == None: return None
v.vtype = JSON_ARRAY
return v
def object(pool: rawptr[memhub.MemManager]) -> rawptr[JsonValue]:
v: rawptr[JsonValue] = JsonValue(pool)
def object(pool: rawptr[memhub.MemManager]) -> JsonValue*:
v: JsonValue* = JsonValue(pool)
if v == None: return None
v.vtype = JSON_OBJECT
return v
# 向数组追加元素
def array_append(pool: rawptr[memhub.MemManager], arr: rawptr[JsonValue], item: rawptr[JsonValue]):
def array_append(pool: rawptr[memhub.MemManager], arr: JsonValue*, item: JsonValue*):
if arr == None or item == None: return
if arr.vtype != JSON_ARRAY: return
arr._append_child(item)
# 向对象设置键值对
def object_set(pool: rawptr[memhub.MemManager], obj: rawptr[JsonValue], key: str, val: rawptr[JsonValue]):
def object_set(pool: rawptr[memhub.MemManager], obj: JsonValue*, key: str, val: JsonValue*):
if obj == None or key == None or val == None: return
if obj.vtype != JSON_OBJECT: return
cur: rawptr[JsonValue] = obj.child
cur: JsonValue* = obj.child
while cur != None:
if cur.key != None:
if string.strcmp(cur.key, key) == 0:
+6 -6
View File
@@ -109,7 +109,7 @@ def _cos_core(x: double) -> double:
return result
def _mul_add(a: uint64, b: uint64, c: uint64, lo: rawptr[uint64], hi: rawptr[uint64]):
def _mul_add(a: uint64, b: uint64, c: uint64, lo: uint64*, hi: uint64*):
"""Write (hi:lo) = a*b + c using 32-bit limbs for the full 128-bit product."""
a0: uint64 = a & 0xFFFFFFFF
a1: uint64 = a >> 32
@@ -128,7 +128,7 @@ def _mul_add(a: uint64, b: uint64, c: uint64, lo: rawptr[uint64], hi: rawptr[uin
*hi = h
def _two_prod(a: double, b: double, ph: rawptr[double], pl: rawptr[double]):
def _two_prod(a: double, b: double, ph: double*, pl: double*):
"""Write a*b as the exact double-double (ph + pl) via Dekker splitting (no FMA required)."""
split: double = 134217729.0
p: double = a * b
@@ -142,7 +142,7 @@ def _two_prod(a: double, b: double, ph: rawptr[double], pl: rawptr[double]):
*pl = ((ahi * bhi - p) + ahi * blo + alo * bhi) + alo * blo
def _rem_pio2(x: double, rh: rawptr[double], rl: rawptr[double]) -> int:
def _rem_pio2(x: double, rh: double*, rl: double*) -> int:
"""O(1) reduction of finite x >= 0: x = n*(pi/2) + r with |r| <= pi/4; returns n mod 4, writes r = rh + rl."""
if x < U_M_REDUCE_LIMIT:
# Cody-Waite three-step with pi/2 = HI + MID + LO; each fn*HK product is exact for fn < 2^20
@@ -393,12 +393,12 @@ def sqrt(x: double) -> double:
if x == 0.0: return 0.0
guess: double = x / 2.0
for _ in range(10):
guess = (guess + x / guess) / double(2.0)
guess = (guess + x / guess) / 2.0
return guess
def abs(x: int) -> int: return x if x >= 0 else -x
def fabs(x: double) -> double: return x if x >= double(0.0) else -x
def fabs(x: double) -> double: return x if x >= 0.0 else -x
def labs(x: long) -> long: return x if x >= 0 else -x
def factorial(n: int) -> double:
@@ -470,7 +470,7 @@ def fmodf(x: float, y: float) -> float:
return result
def modf(x: double, iptr: rawptr[double]) -> double:
def modf(x: double, iptr: double*) -> double:
int_part: double = floor(x)
*iptr = int_part
return x - int_part
+146 -175
View File
@@ -1,10 +1,8 @@
import t, c
from stdint import *
import viperio
import stdlib
import string
import atom
import viperio
import stdio
import stdlib
# ============================================================
# memhub - Unified memory management library
@@ -79,7 +77,7 @@ _memdbg_enable: int = MEMBUDDY_DBG
# set this to 1.
define MEMBUDDY_THREAD_SAFE = 0
# Block-tail canary value (0xA5A5A5A5A5A5A5A5, 64-bit). A t.CDefine makes every use an immediate
# Block-tail canary value (0xA5A5A5A5A5A5A5A5, 64-bit). A define makes every use an immediate
# (the old _canary_pattern() call ran ~26.58M times). Built by shifting (lo << 32) | lo to avoid
# writing the large decimal literal this repo has seen truncated.
define MEMBUDDY_CANARY_PATTERN = (0xA5A5A5A5 << 32) | 0xA5A5A5A5
@@ -146,11 +144,11 @@ def _det_site_put(idx: sizet, ptr: uint64, tag: uint64):
new_cap = idx + 1
if new_cap < 64:
new_cap = 64
new_ptr: t.CPtr = None
new_ptr: rawptr = None
if _det_site_arena is None:
new_ptr = t.CPtr(stdlib.malloc(new_cap * 16))
new_ptr = stdlib.malloc(new_cap * 16)
else:
new_ptr = t.CPtr(stdlib.realloc(t.CPtr(_det_site_arena), new_cap * 16))
new_ptr = stdlib.realloc(rawptr(_det_site_arena), new_cap * 16)
if new_ptr is None:
return
_det_site_arena = (uint64*)(new_ptr)
@@ -161,49 +159,30 @@ def _det_site_put(idx: sizet, ptr: uint64, tag: uint64):
def _det_site_get(idx: sizet, ptr: uint64) -> uint64:
"""Tag of record idx; out of range or ptr mismatch (shifted / cross-instance) returns 0."""
if _det_site_arena is None or idx >= _det_site_cap:
return 0
if uint64(_det_site_arena[idx * 2]) != ptr:
return 0
if _det_site_arena is None or idx >= _det_site_cap: return 0
if uint64(_det_site_arena[idx * 2]) != ptr: return 0
return uint64(_det_site_arena[idx * 2 + 1])
def _det_site_name(tag: uint64) -> str:
"""tag -> "file:line" registry; unregistered tags (including 0) return "?"."""
if tag == uint64(DETSITE_STRUCTGENMIXIN_570):
return "StructGenMixin.vp:570"
if tag == uint64(DETSITE_STRUCTGENMIXIN_698):
return "StructGenMixin.vp:698"
if tag == uint64(DETSITE_STRUCTGENMIXIN_1804):
return "StructGenMixin.vp:1804"
if tag == uint64(DETSITE_CALLGENCTOR_1175):
return "CallGenCtor.vp:1175"
if tag == uint64(DETSITE_CALLGENCTOR_1186):
return "CallGenCtor.vp:1186"
if tag == uint64(DETSITE_STRUCTGENMIXIN_641):
return "StructGenMixin.vp:641"
if tag == uint64(DETSITE_STRUCTGENMIXIN_659):
return "StructGenMixin.vp:659"
if tag == uint64(DETSITE_STRUCTGENMIXIN_693):
return "StructGenMixin.vp:693"
if tag == uint64(DETSITE_LIST_43):
return "_list.vp:43"
if tag == uint64(DETSITE_STRUCTGENMIXIN_372):
return "StructGenMixin.vp:372"
if tag == uint64(DETSITE_STRUCTGENMIXIN_374):
return "StructGenMixin.vp:374"
if tag == uint64(DETSITE_LEAFKIT_1013):
return "CallGenLeafKit.vp:1013"
if tag == uint64(DETSITE_VARGEN_27):
return "VarGenMixin.vp:27"
if tag == uint64(DETSITE_VPIREADER_2626):
return "VpiReader.vp:2626"
if tag == uint64(DETSITE_VPIREADER_2627):
return "VpiReader.vp:2627"
if tag == uint64(DETSITE_CLASSGENDEF_153):
return "ClassGenDef.vp:153"
if tag == uint64(DETSITE_CLASSGENVTABLE_91):
return "ClassGenVTable.vp:91"
if tag == uint64(DETSITE_STRUCTGENMIXIN_570): return "StructGenMixin.vp:570"
if tag == uint64(DETSITE_STRUCTGENMIXIN_698): return "StructGenMixin.vp:698"
if tag == uint64(DETSITE_STRUCTGENMIXIN_1804): return "StructGenMixin.vp:1804"
if tag == uint64(DETSITE_CALLGENCTOR_1175): return "CallGenCtor.vp:1175"
if tag == uint64(DETSITE_CALLGENCTOR_1186): return "CallGenCtor.vp:1186"
if tag == uint64(DETSITE_STRUCTGENMIXIN_641): return "StructGenMixin.vp:641"
if tag == uint64(DETSITE_STRUCTGENMIXIN_659): return "StructGenMixin.vp:659"
if tag == uint64(DETSITE_STRUCTGENMIXIN_693): return "StructGenMixin.vp:693"
if tag == uint64(DETSITE_LIST_43): return "_list.vp:43"
if tag == uint64(DETSITE_STRUCTGENMIXIN_372): return "StructGenMixin.vp:372"
if tag == uint64(DETSITE_STRUCTGENMIXIN_374): return "StructGenMixin.vp:374"
if tag == uint64(DETSITE_LEAFKIT_1013): return "CallGenLeafKit.vp:1013"
if tag == uint64(DETSITE_VARGEN_27): return "VarGenMixin.vp:27"
if tag == uint64(DETSITE_VPIREADER_2626): return "VpiReader.vp:2626"
if tag == uint64(DETSITE_VPIREADER_2627): return "VpiReader.vp:2627"
if tag == uint64(DETSITE_CLASSGENDEF_153): return "ClassGenDef.vp:153"
if tag == uint64(DETSITE_CLASSGENVTABLE_91): return "ClassGenVTable.vp:91"
return "?"
@@ -269,15 +248,13 @@ def SetAllocCtx(stage: bytes, detail: bytes = None):
def _alloc_ctx_stage_str() -> bytes:
"""Current stage name; "?" when unset."""
if _alloc_ctx_stage is None:
return "?"
if _alloc_ctx_stage is None: return "?"
return _alloc_ctx_stage
def _alloc_ctx_detail_str() -> bytes:
"""Current file/module name; "?" when unset."""
if _alloc_ctx_detail is None:
return "?"
if _alloc_ctx_detail is None: return "?"
return _alloc_ctx_detail
@@ -308,33 +285,30 @@ def _block_size_at_order(order: int) -> sizet:
Edge: order <= 0 returns the minimum block (identical to the old empty range loop) and avoids
the undefined negative shift.
"""
if order <= 0:
return MEMBUDDY_MIN_BLOCK
return MEMBUDDY_MIN_BLOCK << sizet(order)
return MEMBUDDY_MIN_BLOCK << (0 if order <= 0 else sizet(order))
# ============================================================
# MemManager - Polymorphic base class
# @t.CVTable explicit annotation to ensure recognition as vtable class during cross-module imports
# virtual explicit annotation to ensure recognition as vtable class during cross-module imports
# (HandlesImports only detects IsCVTable via decorator; automatic inference is unavailable across modules)
# ============================================================
@t.CVTable
class MemManager:
virtual class MemManager:
__provides__: list[str] = ['__memmgr__']
base: t.CPtr # Starting address of memory region
base: rawptr # Starting address of memory region
size: sizet # Total size of memory region
def __init__(self, base: t.CPtr, size: sizet):
def __init__(self, base: rawptr, size: sizet):
self.base = base
self.size = size
# === Virtual functions (overridden by subclasses) ===
def alloc(self, size: sizet) -> t.CPtr:
def alloc(self, size: sizet) -> rawptr:
# Default: allocation not supported
return None
def free(self, ptr: t.CPtr) -> int:
def free(self, ptr: rawptr) -> int:
# Default: no operation
return 0
@@ -344,39 +318,37 @@ class MemManager:
# === Concrete methods (invoke virtual functions for automatic polymorphic dispatch) ===
def calloc(self, count: sizet, size: sizet) -> t.CPtr:
def calloc(self, count: sizet, size: sizet) -> rawptr:
total: sizet = count * size
ptr: t.CPtr = self.alloc(total)
ptr: rawptr = self.alloc(total)
if ptr is not None:
string.memset(ptr, 0, total)
return ptr
def realloc(self, ptr: t.CPtr, new_size: sizet) -> t.CPtr:
def realloc(self, ptr: rawptr, new_size: sizet) -> rawptr:
if ptr is None:
return self.alloc(new_size)
if new_size == 0:
self.free(ptr)
return None
new_ptr: t.CPtr = self.alloc(new_size)
new_ptr: rawptr = self.alloc(new_size)
if new_ptr is None:
return ptr
# Base class cannot track original allocation size; data will not be copied. Subclasses may override to preserve data
self.free(ptr)
return new_ptr
def __enter__(self) -> 'MemManager' | t.CPtr:
def __enter__(self) -> MemManager*:
return self
def __exit__(self):
self.reset()
def alloc_buf(self, capacity: sizet) -> rawptr[viperio.Buf]:
buf: rawptr[viperio.Buf] = self.alloc(viperio.Buf.__sizeof__())
if buf is None:
return None
data_ptr: t.CPtr = self.alloc(capacity)
if data_ptr is None:
return None
def alloc_buf(self, capacity: sizet) -> viperio.Buf*:
buf: viperio.Buf* = self.alloc(viperio.Buf.__sizeof__())
if buf is None: return None
data_ptr: rawptr = self.alloc(capacity)
if data_ptr is None: return None
buf.__before_init__()
buf.__init__(data_ptr, capacity)
return buf
@@ -390,7 +362,7 @@ class MemPool(MemManager):
offset: sizet # Bump pointer offset
high_water: sizet # High-water mark
def __init__(self, base: t.CPtr, size: sizet):
def __init__(self, base: rawptr, size: sizet):
self.base = base
self.size = size
self.offset = 0
@@ -417,7 +389,7 @@ class MemPool(MemManager):
# * this round cannot compile/run to add per-site None checks.
# New callers needing a controllable failure should pre-check with AllocOk(). The None contract
# is provided by MemBuddy; MemPool's contract stays "never returns None, terminates on exhaustion".
def alloc(self, size: sizet) -> t.CPtr:
def alloc(self, size: sizet) -> rawptr:
if size == 0: return None
aligned: sizet = _align_up(size, MEMHUB_ALIGN)
if self.offset + aligned > self.size:
@@ -428,13 +400,13 @@ class MemPool(MemManager):
stdio.printf("[OOM] 说明: 内存池已耗尽。可能原因: (1) 池大小不足,需增大 POOL_SIZE; (2) 存在内存泄漏。\n")
stdio.printf("[OOM] 提示: 需要可控失败而非终止的调用方,请先用 MemPool.AllocOk(size) 预检。\n")
stdlib.exit(1)
ptr: t.CPtr = self.base + self.offset
ptr: rawptr = self.base + self.offset
self.offset += aligned
if self.offset > self.high_water:
self.high_water = self.offset
return ptr
def free(self, ptr: t.CPtr) -> int:
def free(self, ptr: rawptr) -> int:
# Bump allocator does not support individual free operations
return 1
@@ -468,10 +440,9 @@ class MemPool(MemManager):
# Construct a MemPool (arena bump allocator) on the heap. base = arena start, size = its byte count.
# Returns None when malloc fails.
def NewMemPool(base: t.CPtr, size: sizet) -> rawptr[MemPool]:
obj: rawptr[MemPool] = stdlib.malloc(MemPool.__sizeof__())
if obj is None:
return None
def NewMemPool(base: rawptr, size: sizet) -> MemPool*:
obj: MemPool* = stdlib.malloc(MemPool.__sizeof__())
if obj is None: return None
obj.__before_init__()
obj.__init__(base, size)
return obj
@@ -482,11 +453,11 @@ def NewMemPool(base: t.CPtr, size: sizet) -> rawptr[MemPool]:
# detection) and then calls the existing virtual alloc => no new virtual function, no vtable change,
# no contract change; for a MemBuddy pool its alloc_tagged is also available (that one additionally
# records a record-table tag for [FREEFRAME] reports).
def MemAllocTagged(pool: rawptr[MemManager], size: sizet, tag: sizet) -> t.CPtr:
def MemAllocTagged(pool: rawptr[MemManager], size: sizet, tag: sizet) -> rawptr:
if pool is None:
return None
prev: uint64 = _alloc_site_enter(uint64(tag))
r: t.CPtr = pool.alloc(size)
r: rawptr = pool.alloc(size)
_alloc_site_leave(prev)
return r
@@ -499,13 +470,13 @@ class MemSlab(MemManager):
block_size: sizet # Block size (aligned)
block_count: sizet # Total block count
used_count: sizet # Used block count
free_list: t.CPtr # Free list head
free_list: rawptr # Free list head
alloc_map: uint8* # Allocation bitmap (at start of arena)
alloc_map_size: sizet # Bitmap byte count
usable: t.CPtr # Usable block area start (after bitmap)
usable: rawptr # Usable block area start (after bitmap)
usable_size: sizet # Usable block area size
def __init__(self, base: t.CPtr, size: sizet, block_size: sizet):
def __init__(self, base: rawptr, size: sizet, block_size: sizet):
self.base = base
self.size = size
self.block_size = 0
@@ -544,12 +515,12 @@ class MemSlab(MemManager):
self.free_list = None
i: sizet = 0
while i < self.block_count:
block: t.CPtr = self.usable + i * bs
block: rawptr = self.usable + i * bs
*((uint64*)(block)) = uint64(self.free_list)
self.free_list = block
i += 1
def alloc(self, size: sizet) -> t.CPtr:
def alloc(self, size: sizet) -> rawptr:
# Slab mode: allocate one block only if requested size <= block_size
if self.free_list is None:
stdio.printf("\n[OOM] 内存不足 (Out of Memory)\n")
@@ -564,14 +535,14 @@ class MemSlab(MemManager):
stdio.printf("[WARN] MemSlab: 请求 %zu 字节超过 block_size=%zu,分配失败返回 NULL;应改用 MemBuddy/MemPool。\n",
size, self.block_size)
return None
block: t.CPtr = self.free_list
self.free_list = t.CPtr(*((uint64*)(block)))
block: rawptr = self.free_list
self.free_list = rawptr(*((uint64*)(block)))
self.used_count += 1
idx: sizet = (uint64(block) - uint64(self.usable)) / self.block_size
self.alloc_map[idx / 8] = self.alloc_map[idx / 8] | uint8(1 << (idx % 8))
return block
def free(self, ptr: t.CPtr) -> int:
def free(self, ptr: rawptr) -> int:
if ptr is None: return 0
p: uint64 = uint64(ptr)
if p < uint64(self.usable): return 0
@@ -592,7 +563,7 @@ class MemSlab(MemManager):
self.free_list = None
i: sizet = 0
while i < self.block_count:
block: t.CPtr = self.usable + i * self.block_size
block: rawptr = self.usable + i * self.block_size
*((uint64*)(block)) = uint64(self.free_list)
self.free_list = block
if i / 8 < self.alloc_map_size:
@@ -619,13 +590,13 @@ class MemBuddy(MemManager):
__provides__: list[str] = ['__mbuddy__', '__memmgr__']
max_order: int # maximum order
free_lists: uint64* # free-list head array (at the arena start)
lock_val: t.CVolatile | int # spinlock
usable: t.CPtr # start of usable memory (skips free_lists)
lock_val: volatile int # spinlock
usable: rawptr # start of usable memory (skips free_lists)
usable_size: sizet # usable memory size (sum of decomposed blocks)
chk_stack: uint64* # checkpoint record stack (stores block addresses)
chk_cap: sizet # record-stack capacity (entries)
chk_depth: sizet # current record depth
chk_base: t.CPtr # record-stack base (stdlib.malloc, separate from the buddy area)
chk_base: rawptr # record-stack base (stdlib.malloc, separate from the buddy area)
big_alloc_count: sizet # count of allocations >= 8KB (OOM diagnostics)
# mark() 时刻的空闲字节(OOM 诊断):据此区分「本窗口把池吃满」与「进窗口前就已满」
mark_free_bytes: sizet = 0
@@ -639,12 +610,12 @@ class MemBuddy(MemManager):
# full-table scan. Diagnostics only, very slow. Default 0 = identical to the historical build.
deep_detect: int = 0
detection_size: sizet = MEMBUDDY_DET_RECORD_SIZE
_det_arena: t.CPtr = None # record-area base (stdlib.malloc, for realloc growth)
_det_arena: rawptr = None # record-area base (stdlib.malloc, for realloc growth)
_det_records: uint64* = None # record table (each 2×u64 = {ptr, size})
_det_count: sizet = 0 # number of records
_det_cap: sizet = 0 # record capacity (entries)
def __init__(self, base: t.CPtr, size: sizet,
def __init__(self, base: rawptr, size: sizet,
detection: int = 0, deep_detect: int = 0):
self.base = base
self.size = size
@@ -731,10 +702,10 @@ class MemBuddy(MemManager):
# Records + overlap detection are decoupled from detection: even without it the record area
# is allocated and same-address duplicate-allocation detection is on (no markers, no layout
# change), which locates free-list / duplicate-allocation corruption.
det_arena: t.CPtr = t.CPtr(stdlib.malloc(self.detection_size))
det_arena: rawptr = stdlib.malloc(self.detection_size)
if det_arena is not None:
self._det_arena = det_arena
self._det_records = (uint64*)(det_arena)
self._det_records = uint64*(det_arena)
self._det_cap = self.detection_size // 16
self._det_count = 0
@@ -752,7 +723,7 @@ class MemBuddy(MemManager):
"""
if self.chk_stack is None:
cap: sizet = 2 * 1024 * 1024
base: t.CPtr = t.CPtr(stdlib.malloc(cap * 8))
base: rawptr = stdlib.malloc(cap * 8)
if base is None:
return 0
self.chk_base = base
@@ -786,10 +757,10 @@ class MemBuddy(MemManager):
self._lock()
while self.chk_depth > mark:
self.chk_depth -= 1
block: t.CPtr = t.CPtr(self.chk_stack[self.chk_depth])
block: rawptr = rawptr(self.chk_stack[self.chk_depth])
if block is None:
continue
stored: t.CPtr = t.CPtr(*((uint64*)(block)))
stored: rawptr = rawptr(*((uint64*)(block)))
stored_val: uint64 = uint64(stored)
if (stored_val & 1) == 0:
continue # already freed individually: skip
@@ -817,7 +788,7 @@ class MemBuddy(MemManager):
# === Free list operations ===
def _fl_push(self, order: int, block: t.CPtr):
def _fl_push(self, order: int, block: rawptr):
old_head: uint64 = self.free_lists[order]
*((uint64*)(block)) = old_head
self.free_lists[order] = uint64(block)
@@ -828,7 +799,7 @@ class MemBuddy(MemManager):
if self.detection != 0:
self._ff_push_mark(order, block, old_head)
def _fl_block_ok(self, block: t.CPtr) -> int:
def _fl_block_ok(self, block: rawptr) -> int:
"""Validate a free-list block address: it must lie in [usable, usable+usable_size) and
be aligned to MEMBUDDY_MIN_BLOCK (same range+alignment criteria as self_check).
@@ -850,11 +821,11 @@ class MemBuddy(MemManager):
# === [FREEFRAME] free-block stamps: forensics (read/write diagnostic words; detection only) ===
def _ff_word(self, block: t.CPtr, idx: int) -> uint64:
def _ff_word(self, block: rawptr, idx: int) -> uint64:
"""Read word idx of a free block (0=next, 1=order stamp, 2=check word, 3=next mirror)."""
return (uint64*)(block)[idx]
def _ff_word_set(self, block: t.CPtr, idx: int, val: uint64):
def _ff_word_set(self, block: rawptr, idx: int, val: uint64):
"""Write word idx of a free block (diagnostic words only; word0 is never touched)."""
p: uint64* = (uint64*)(block)
p[idx] = val
@@ -863,7 +834,7 @@ class MemBuddy(MemManager):
"""Order stamp of a free block of `order` = TAG ^ (order << 32) (order is recoverable)."""
return uint64(MEMBUDDY_FREEFRAME_TAG) ^ (uint64(order) << 32)
def _ff_push_mark(self, order: int, block: t.CPtr, next_val: uint64):
def _ff_push_mark(self, order: int, block: rawptr, next_val: uint64):
"""Stamp words 1..3 after linking. The caller has already checked detection != 0 and
written word0 = next_val.
@@ -876,7 +847,7 @@ class MemBuddy(MemManager):
self._ff_word_set(block, 2, uint64(MEMBUDDY_FREEFRAME_TAG2))
self._ff_word_set(block, 3, next_val)
def _ff_report(self, order: int, block: t.CPtr):
def _ff_report(self, order: int, block: rawptr):
"""[FREEFRAME] forensic dump (the caller then fflush + exit(1)).
Three parts that together name the writer:
@@ -975,7 +946,7 @@ class MemBuddy(MemManager):
# [FREEFRAME] Dump the writer's payload head/tail to separate an object
# overflow from a wild pointer landing on this block's tail (see
# _ff_owner_dump).
self._ff_owner_dump(t.CPtr(l_addr), o)
self._ff_owner_dump(rawptr(l_addr), o)
# [FREEFRAME] Evidence-based left-neighbour attribution (geometry plus
# canary alone cannot attribute; criteria in _ff_left_classify).
self._ff_left_classify(l_addr, bs_o, lcan, w0)
@@ -1007,7 +978,7 @@ class MemBuddy(MemManager):
stdio.printf("\n")
stdio.fflush(None)
def _ff_owner_dump(self, block: t.CPtr, order: int):
def _ff_owner_dump(self, block: rawptr, order: int):
"""[FREEFRAME] Dump the writer (allocation) block's payload head/tail (16 bytes each) and
its requested size.
@@ -1109,8 +1080,8 @@ class MemBuddy(MemManager):
nw: sizet = (l_bs - MEMBUDDY_HEADER_SIZE - MEMBUDDY_CANARY) / 8
tail_zero: int = 1
if nw >= 2:
w_a: uint64 = self._ff_word(t.CPtr(l_addr), int(nw) - 1)
w_b: uint64 = self._ff_word(t.CPtr(l_addr), int(nw))
w_a: uint64 = self._ff_word(rawptr(l_addr), int(nw) - 1)
w_b: uint64 = self._ff_word(rawptr(l_addr), int(nw))
if w_a != 0 or w_b != 0:
tail_zero = 0
hit_v: sizet = 0
@@ -1118,7 +1089,7 @@ class MemBuddy(MemManager):
out_pool: sizet = 0
i: sizet = 0
while i < nw:
wv: uint64 = self._ff_word(t.CPtr(l_addr), int(i) + 1)
wv: uint64 = self._ff_word(rawptr(l_addr), int(i) + 1)
if wv == lcan:
hit_v += 1
if wv != 0:
@@ -1146,7 +1117,7 @@ class MemBuddy(MemManager):
if case_n == 0:
stdio.printf("[FREEFRAME] ⇒ 结论C:可区分特征均未命中 ⇒ 不据「几何 + canary 损坏」断言归属(需结合记录表现场与 payload 全量转储另证)\n")
def _ff_verify(self, order: int, block: t.CPtr):
def _ff_verify(self, order: int, block: rawptr):
"""[FREEFRAME] Verify the free-block stamps before popping (detection only, caller-checked).
A stamp/mirror mismatch means the block's words were written off-list after linking, so
@@ -1173,7 +1144,7 @@ class MemBuddy(MemManager):
stdio.fflush(None)
stdlib.exit(1)
def _ff_owner_check(self, block: t.CPtr, order: int):
def _ff_owner_check(self, block: rawptr, order: int):
"""[FREEFRAME] Pre-free owner check (detection only): verify the block-tail canary.
A contiguous overflow writer always overwrites its own block-tail canary before entering
@@ -1221,7 +1192,7 @@ class MemBuddy(MemManager):
stdio.fflush(None)
stdlib.exit(1)
def _fl_pop(self, order: int) -> t.CPtr:
def _fl_pop(self, order: int) -> rawptr:
"""Pop the head of order `order`'s free list (O(1): no loop/recursion in this body).
Value validation (fail-fast): this function has only two memory accesses -- read
@@ -1251,21 +1222,21 @@ class MemBuddy(MemManager):
head_val: uint64 = self.free_lists[order]
if head_val == 0:
return None
if self._fl_block_ok(t.CPtr(head_val)) == 0:
if self._fl_block_ok(rawptr(head_val)) == 0:
stdio.printf("[MEM] _fl_pop 调用者=_split_to_order 链表头非法: order=%d head=%llu max_order=%d usable_size=%zu(free_lists[order] 被写穿:非池内对齐块地址)\n",
order, head_val, self.max_order, self.usable_size)
stdio.fflush(None)
stdlib.exit(1)
block: t.CPtr = t.CPtr(head_val)
block: rawptr = rawptr(head_val)
# [FREEFRAME] detection mode: verify the free-block stamps/mirror before popping (earlier
# and more sensitive than the "is next a valid block address" guard below, which a writer
# that happened to write a valid 8-byte block address would pass). With detection == 0 it
# is one predictable non-taken branch, no extra memory access.
if self.detection != 0:
self._ff_verify(order, block)
next_ptr: t.CPtr = t.CPtr(*((uint64*)(block)))
next_ptr: rawptr = rawptr(*((uint64*)(block)))
next_val: uint64 = uint64(next_ptr)
if next_val != 0 and self._fl_block_ok(t.CPtr(next_val)) == 0:
if next_val != 0 and self._fl_block_ok(rawptr(next_val)) == 0:
stdio.printf("[MEM] _fl_pop 调用者=_split_to_order next 指针非法: order=%d block=%llu next=%llu max_order=%d usable_size=%zu(空闲块首字被写穿:next 非池内对齐块地址)\n",
order, head_val, next_val, self.max_order, self.usable_size)
stdio.fflush(None)
@@ -1277,11 +1248,11 @@ class MemBuddy(MemManager):
self.free_lists[order] = next_val
return block
def _fl_find_and_remove(self, order: int, target: t.CPtr) -> int:
def _fl_find_and_remove(self, order: int, target: rawptr) -> int:
head_val: uint64 = self.free_lists[order]
if head_val == 0:
return 0
head: t.CPtr = t.CPtr(head_val)
head: rawptr = rawptr(head_val)
# [FREEFRAME] detection mode: this function walks the chain (following next), so a
# written-through node would send the walk into a wild chain (a cycle/over-limit report
# only says "heap is broken", naming nothing). Verify the head and every visited node, so
@@ -1289,11 +1260,11 @@ class MemBuddy(MemManager):
if self.detection != 0:
self._ff_verify(order, head)
if uint64(head) == uint64(target):
next_ptr: t.CPtr = t.CPtr(*((uint64*)(head)))
next_ptr: rawptr = rawptr(*((uint64*)(head)))
self.free_lists[order] = uint64(next_ptr)
return 1
prev: t.CPtr = head
cur: t.CPtr = t.CPtr(*((uint64*)(head)))
prev: rawptr = head
cur: rawptr = rawptr(*((uint64*)(head)))
# Cycle guard (same bound as self_check): a corrupted next pointer can make the free list
# cyclic, and without a step bound this loop would never end (free/coalesce hang with a
# drifting crash). The bound is "all minimum blocks + 2", which no valid free count can
@@ -1313,7 +1284,7 @@ class MemBuddy(MemManager):
if self.detection != 0:
self._ff_verify(order, cur)
if uint64(cur) == uint64(target):
next_ptr: t.CPtr = t.CPtr(*((uint64*)(cur)))
next_ptr: rawptr = rawptr(*((uint64*)(cur)))
*((uint64*)(prev)) = uint64(next_ptr)
# [FREEFRAME] Middle unlink rewrote prev's word0 (next), so its mirror word3 must
# be synced too; otherwise _ff_verify would misreport memhub's own valid unlink as
@@ -1323,12 +1294,12 @@ class MemBuddy(MemManager):
self._ff_word_set(prev, 3, uint64(next_ptr))
return 1
prev = cur
cur = t.CPtr(*((uint64*)(cur)))
cur = rawptr(*((uint64*)(cur)))
return 0
# === Buddy system core ===
def _buddy_of(self, block: t.CPtr, order: int) -> t.CPtr:
def _buddy_of(self, block: rawptr, order: int) -> rawptr:
# Buddy = block with the order's size bit flipped in its pool byte offset.
offset: sizet = uint64(block) - uint64(self.usable)
bs: sizet = MEMBUDDY_MIN_BLOCK << sizet(order)
@@ -1352,7 +1323,7 @@ class MemBuddy(MemManager):
order += 1
return order
def _split_to_order(self, to_order: int) -> t.CPtr:
def _split_to_order(self, to_order: int) -> rawptr:
# max_order must be in range before it indexes free_lists (a corrupt field would
# otherwise read out of bounds and feed garbage to _fl_pop). fail-fast.
if self.max_order < 0 or self.max_order > MEMBUDDY_MAX_ORDERS:
@@ -1368,16 +1339,16 @@ class MemBuddy(MemManager):
found_order += 1
if found_order > self.max_order:
return None
block: t.CPtr = self._fl_pop(found_order)
block: rawptr = self._fl_pop(found_order)
while found_order > to_order:
# Split the block in half, pushing the upper half back onto the free list.
found_order -= 1
bs: sizet = MEMBUDDY_MIN_BLOCK << sizet(found_order)
buddy: t.CPtr = block + bs
buddy: rawptr = block + bs
self._fl_push(found_order, buddy)
return block
def _coalesce(self, block: t.CPtr, order: int):
def _coalesce(self, block: rawptr, order: int):
# Value validation (fail-fast, same rule as _fl_pop/_split_to_order): the first loop
# indexes self.free_lists[order] directly, and order comes from the header `stored >> 1`
# (free/reset_to paths) -- a written-through header yields an arbitrary index => out-of-
@@ -1398,7 +1369,7 @@ class MemBuddy(MemManager):
# has side effects, so the free-list semantics and order are fully equivalent.
if self.free_lists[order] == 0:
break
buddy: t.CPtr = self._buddy_of(block, order)
buddy: rawptr = self._buddy_of(block, order)
found: int = self._fl_find_and_remove(order, buddy)
if found == 0:
break
@@ -1407,15 +1378,15 @@ class MemBuddy(MemManager):
order += 1
self._fl_push(order, block)
def _is_valid_ptr(self, ptr: t.CPtr) -> int:
def _is_valid_ptr(self, ptr: rawptr) -> int:
if ptr is None: return 0
if self.usable is None: return 0
block: t.CPtr = ptr - MEMBUDDY_HEADER_SIZE
block: rawptr = ptr - MEMBUDDY_HEADER_SIZE
if uint64(block) < uint64(self.usable): return 0
if uint64(block) >= uint64(self.usable) + self.usable_size: return 0
offset: sizet = uint64(block) - uint64(self.usable)
if offset % MEMBUDDY_MIN_BLOCK != 0: return 0
stored: t.CPtr = t.CPtr(*((uint64*)(block)))
stored: rawptr = rawptr(*((uint64*)(block)))
stored_val: uint64 = uint64(stored)
if (stored_val & 1) == 0: return 0
order: int = int(stored_val >> 1)
@@ -1451,9 +1422,9 @@ class MemBuddy(MemManager):
atom.__atomic_clear(&self.lock_val, atom.ATOMIC_RELEASE)
# === Virtual function overrides ===
def alloc(self, size: sizet) -> t.CPtr:
def alloc(self, size: sizet) -> rawptr:
self._lock()
result: t.CPtr = None
result: rawptr = None
if size != 0 and self.usable is not None:
# OOM 诊断:≥8KB 的请求计数。此字段此前从不自增,报告里恒为 big=0,
# 无法据此分辨“多次小块”与“少量大块”。阈值与字段注释口径一致。
@@ -1477,7 +1448,7 @@ class MemBuddy(MemManager):
stdlib.exit(1)
order: int = self._order_for_size(needed)
if order <= self.max_order:
block: t.CPtr = self._split_to_order(order)
block: rawptr = self._split_to_order(order)
if block is not None:
*((uint64*)(block)) = uint64((order << 1) | 1) # header: odd = allocated
result = block + MEMBUDDY_HEADER_SIZE
@@ -1502,7 +1473,7 @@ class MemBuddy(MemManager):
if self.chk_stack is not None:
if self.chk_depth >= self.chk_cap:
new_cap: sizet = self.chk_cap * 2
new_stack: t.CPtr = t.CPtr(stdlib.realloc(self.chk_base, new_cap * 8))
new_stack: rawptr = rawptr(stdlib.realloc(self.chk_base, new_cap * 8))
if new_stack is not None:
self.chk_base = new_stack
self.chk_stack = (uint64*)(new_stack)
@@ -1552,8 +1523,8 @@ class MemBuddy(MemManager):
if _memdbg_enable != 0:
# [MEMDBG] A freshly popped block must not still be on its own free
# list (a duplicate put would leave a residual node -> reallocated).
dbg_block: t.CPtr = result - MEMBUDDY_HEADER_SIZE
dbg_hdr: t.CPtr = t.CPtr(*((uint64*)(dbg_block)))
dbg_block: rawptr = result - MEMBUDDY_HEADER_SIZE
dbg_hdr: rawptr = rawptr(*((uint64*)(dbg_block)))
dbg_order: int = int(uint64(dbg_hdr) >> 1)
# dbg_order comes from the block header; if the header were corrupted
# (the very case this branch hunts), indexing free_lists[dbg_order]
@@ -1564,7 +1535,7 @@ class MemBuddy(MemManager):
uint64(dbg_block))
stdio.fflush(None)
stdlib.exit(1)
dbg_cur: t.CPtr = t.CPtr(self.free_lists[dbg_order])
dbg_cur: rawptr = rawptr(self.free_lists[dbg_order])
dbg_n: sizet = 0
while uint64(dbg_cur) != 0:
if uint64(dbg_cur) == uint64(dbg_block):
@@ -1577,11 +1548,11 @@ class MemBuddy(MemManager):
stdio.printf("[MEMDBG] alloc 扫描 free list order=%d 成环/超长(自检异常,终止)\n", dbg_order)
stdio.fflush(None)
stdlib.exit(1)
dbg_cur = t.CPtr(*((uint64*)(dbg_cur)))
dbg_cur = rawptr(*((uint64*)(dbg_cur)))
if self._det_count >= self._det_cap:
# Record table full: double it so is_alive/_det_probe never miss live blocks.
new_cap: sizet = self._det_cap * 2
new_rec: t.CPtr = t.CPtr(stdlib.realloc(self._det_arena, new_cap * 16))
new_rec: rawptr = rawptr(stdlib.realloc(self._det_arena, new_cap * 16))
if new_rec is not None:
self._det_arena = new_rec
self._det_records = (uint64*)(new_rec)
@@ -1670,7 +1641,7 @@ class MemBuddy(MemManager):
self._unlock()
return result
def alloc_tagged(self, size: sizet, tag: sizet) -> t.CPtr:
def alloc_tagged(self, size: sizet, tag: sizet) -> rawptr:
"""alloc wrapper that records the caller's site tag for [FREEFRAME] / zero-size reports.
Same semantics as alloc, plus two registrations:
@@ -1683,20 +1654,20 @@ class MemBuddy(MemManager):
"""
prev: uint64 = _alloc_site_enter(uint64(tag))
if self.detection == 0 or self._det_records is None or tag == 0:
r_plain: t.CPtr = self.alloc(size)
r_plain: rawptr = self.alloc(size)
_alloc_site_leave(prev)
return r_plain
before: sizet = self._det_count
r: t.CPtr = self.alloc(size)
r: rawptr = self.alloc(size)
if r is not None and self._det_count == before + 1:
_det_site_put(before, uint64(r), uint64(tag))
_alloc_site_leave(prev)
return r
def free(self, ptr: t.CPtr) -> int:
def free(self, ptr: rawptr) -> int:
self._lock()
if ptr is not None:
p_eff: t.CPtr = ptr
p_eff: rawptr = ptr
if self.detection != 0:
# [DET] Returned ptr is 2 bytes past the user area; step back on free.
p_eff = ptr - 2
@@ -1717,8 +1688,8 @@ class MemBuddy(MemManager):
break
ri += 1
if self._is_valid_ptr(p_eff) != 0:
block: t.CPtr = p_eff - MEMBUDDY_HEADER_SIZE
stored: t.CPtr = t.CPtr(*((uint64*)(block)))
block: rawptr = p_eff - MEMBUDDY_HEADER_SIZE
stored: rawptr = rawptr(*((uint64*)(block)))
stored_val: uint64 = uint64(stored)
order: int = int(stored_val >> 1)
if (stored_val & 1) == 0:
@@ -1740,7 +1711,7 @@ class MemBuddy(MemManager):
if _memdbg_enable != 0:
# [MEMDBG] Double-return check: a live block (odd header) must never
# already be on this order's free list; a hit = the same block put twice.
dbg_f: t.CPtr = t.CPtr(self.free_lists[order])
dbg_f: rawptr = rawptr(self.free_lists[order])
dbg_fn: sizet = 0
while uint64(dbg_f) != 0:
if uint64(dbg_f) == uint64(block):
@@ -1753,7 +1724,7 @@ class MemBuddy(MemManager):
stdio.printf("[MEMDBG] free 扫描 free list order=%d 成环/超长(自检异常,终止)\n", order)
stdio.fflush(None)
stdlib.exit(1)
dbg_f = t.CPtr(*((uint64*)(dbg_f)))
dbg_f = rawptr(*((uint64*)(dbg_f)))
*((uint64*)(block)) = 0
self._coalesce(block, order)
elif _memdbg_enable != 0:
@@ -1854,20 +1825,20 @@ class MemBuddy(MemManager):
return 1
# realloc override: buddy system reads original order from block header; old_size is not required
def realloc(self, ptr: t.CPtr, new_size: sizet) -> t.CPtr:
def realloc(self, ptr: rawptr, new_size: sizet) -> rawptr:
if ptr is None:
return self.alloc(new_size)
if new_size == 0:
self.free(ptr)
return None
p_eff: t.CPtr = ptr
p_eff: rawptr = ptr
if self.detection != 0:
# [DET] Same as free: step back 2 bytes before block-level math.
p_eff = ptr - 2
if self._is_valid_ptr(p_eff) == 0:
return None
block: t.CPtr = p_eff - MEMBUDDY_HEADER_SIZE
stored: t.CPtr = t.CPtr(*((uint64*)(block)))
block: rawptr = p_eff - MEMBUDDY_HEADER_SIZE
stored: rawptr = rawptr(*((uint64*)(block)))
stored_val: uint64 = uint64(stored)
old_order: int = int(stored_val >> 1)
# Capacity must match alloc exactly (single source of truth): alloc's block capacity is
@@ -1881,7 +1852,7 @@ class MemBuddy(MemManager):
new_order: int = self._order_for_size(needed)
if new_order <= old_order:
return ptr
new_ptr: t.CPtr = self.alloc(new_size)
new_ptr: rawptr = self.alloc(new_size)
if new_ptr is None:
return ptr
old_block_size: sizet = _block_size_at_order(old_order)
@@ -1900,7 +1871,7 @@ class MemBuddy(MemManager):
# Covers _det_probe's blind spot: write-side markers cannot catch use-after-free
# reads (a dangling pointer into a reused block).
# ============================================================
def is_alive(self, ptr: t.CPtr) -> int:
def is_alive(self, ptr: rawptr) -> int:
if self.detection == 0:
return 1
if self._det_records is None:
@@ -2013,7 +1984,7 @@ class MemBuddy(MemManager):
max_steps: sizet = self.usable_size / MEMBUDDY_MIN_BLOCK + 2
for i in range(MEMBUDDY_MAX_ORDERS + 1):
head_val: uint64 = self.free_lists[i]
cur: t.CPtr = t.CPtr(head_val)
cur: rawptr = rawptr(head_val)
if uint64(cur) != 0 and self._fl_block_ok(cur) == 0:
# 链表头本身非法(如未初始化 0xBE…):直接解引用会 AV,这里报告后停止本阶。
stdio.printf("[MEM] free_lists[%d] 头非法(非池内对齐块地址,疑元数据未初始化/被写穿): "
@@ -2029,7 +2000,7 @@ class MemBuddy(MemManager):
i, head_val)
break
count += 1
next_p: t.CPtr = t.CPtr(*((uint64*)(cur)))
next_p: rawptr = rawptr(*((uint64*)(cur)))
if uint64(next_p) != 0 and self._fl_block_ok(next_p) == 0:
# 头/节点非法(未初始化 0xBE…/非池内对齐地址):继续解引用会直接 AV。
# 这里只报告并停止本阶计数——"链表头是未初始化内存"本身就是元数据损坏的铁证。
@@ -2050,7 +2021,7 @@ class MemBuddy(MemManager):
max_steps: sizet = self.usable_size / MEMBUDDY_MIN_BLOCK + 2
for i in range(MEMBUDDY_MAX_ORDERS + 1):
head_val: uint64 = self.free_lists[i]
cur: t.CPtr = t.CPtr(head_val)
cur: rawptr = rawptr(head_val)
if uint64(cur) != 0 and self._fl_block_ok(cur) == 0:
# 链表头本身非法(如未初始化 0xBE…):直接解引用会 AV,这里报告后停止本阶。
stdio.printf("[MEM] free_lists[%d] 头非法(非池内对齐块地址,疑元数据未初始化/被写穿): "
@@ -2066,7 +2037,7 @@ class MemBuddy(MemManager):
i, head_val)
break
total += _block_size_at_order(i)
next_b: t.CPtr = t.CPtr(*((uint64*)(cur)))
next_b: rawptr = rawptr(*((uint64*)(cur)))
if uint64(next_b) != 0 and self._fl_block_ok(next_b) == 0:
stdio.printf("[MEM] free_lists[%d] 节点非法(非池内对齐块地址,疑元数据未初始化/被写穿): "
"head=%llu cur=%llu next=%llu chk_depth=%zu usable=%llu usable_size=%zu\n",
@@ -2122,7 +2093,7 @@ class MemBuddy(MemManager):
pool_end: uint64 = uint64(self.usable) + self.usable_size
ci: sizet = 0
while ci < self.chk_depth:
block: t.CPtr = t.CPtr(self.chk_stack[ci])
block: rawptr = rawptr(self.chk_stack[ci])
ci += 1
if block is None:
bad += 1
@@ -2131,7 +2102,7 @@ class MemBuddy(MemManager):
if bv < uint64(self.usable) or bv >= pool_end:
bad += 1
continue
stored: t.CPtr = t.CPtr(*((uint64*)(block)))
stored: rawptr = rawptr(*((uint64*)(block)))
stored_val: uint64 = uint64(stored)
if (stored_val & 1) == 0:
continue
@@ -2189,18 +2160,18 @@ class MemBuddy(MemManager):
hv_m8: uint64 = 0
hv_p8: uint64 = 0
if raw >= uint64(self.usable) and raw < pool_end:
bp: t.CPtr = t.CPtr(raw)
bp: rawptr = rawptr(raw)
# 同址两种读法交叉验证:hv 用直读 `*((uint64*)(p))`,hv2 用 memhub
# 其余路径通用的两步式 `uint64(t.CPtr(*((uint64*)(p))))`。若两者不一致,
# 其余路径通用的两步式 `uint64(rawptr(*((uint64*)(p))))`。若两者不一致,
# 说明「块头读作 0」是读法/代码生成问题,而非池真的被清零。
hv = *((uint64*)(bp))
hv2 = uint64(t.CPtr(*((uint64*)(bp))))
hv2 = uint64(rawptr(*((uint64*)(bp))))
# 邻字探针:整片被零填充 ⇒ 前后邻字同样为 0(一次巨量零填充写的签名)
if raw >= uint64(self.usable) + 8:
bp_m: t.CPtr = t.CPtr(raw - 8)
bp_m: rawptr = rawptr(raw - 8)
hv_m8 = *((uint64*)(bp_m))
if raw + 8 < pool_end:
bp_p: t.CPtr = t.CPtr(raw + 8)
bp_p: rawptr = rawptr(raw + 8)
hv_p8 = *((uint64*)(bp_p))
stdio.printf("[MEM] [%zu] addr=%llu hdr=0x%llX hdr2=0x%llX "
"-8=0x%llX +8=0x%llX\n", si, raw, hv, hv2, hv_m8, hv_p8)
@@ -2219,7 +2190,7 @@ class MemBuddy(MemManager):
i: int
for i in range(MEMBUDDY_MAX_ORDERS + 1):
head_val: uint64 = self.free_lists[i]
cur: t.CPtr = t.CPtr(head_val)
cur: rawptr = rawptr(head_val)
if uint64(cur) != 0 and self._fl_block_ok(cur) == 0:
# 链表头本身非法(如未初始化 0xBE…):直接解引用会 AV,这里报告后停止本阶。
stdio.printf("[MEM] free_lists[%d] 头非法(非池内对齐块地址,疑元数据未初始化/被写穿): "
@@ -2242,7 +2213,7 @@ class MemBuddy(MemManager):
bs: sizet = _block_size_at_order(i)
if offset % bs != 0:
return 3
cur = t.CPtr(*((uint64*)(cur)))
cur = rawptr(*((uint64*)(cur)))
return 0
def check_alive_blocks(self, from_mark: sizet) -> int:
@@ -2265,7 +2236,7 @@ class MemBuddy(MemManager):
return 0
ci: sizet = from_mark
while ci < self.chk_depth:
block: t.CPtr = t.CPtr(self.chk_stack[ci])
block: rawptr = rawptr(self.chk_stack[ci])
if block is not None:
if uint64(block) < uint64(self.usable):
# Block address below the pool start.
@@ -2277,7 +2248,7 @@ class MemBuddy(MemManager):
stdio.printf("[MEM] 存活块损坏 rc=2(块地址超出池末尾): block=%llu usable=%llu ci=%zu from_mark=%zu\n",
uint64(block), uint64(self.usable), ci, from_mark)
return 2
stored: t.CPtr = t.CPtr(*((uint64*)(block)))
stored: rawptr = rawptr(*((uint64*)(block)))
stored_val: uint64 = uint64(stored)
if (stored_val & 1) != 0:
# Live block: header must be (order<<1)|1 with a valid order and alignment.
@@ -2303,7 +2274,7 @@ class MemBuddy(MemManager):
# an 8-byte value, so it must be read through a uint64* view, and the byte
# address is computed on the byte pointer first (arithmetic on a typed pointer
# scales by sizeof(T)=8 and would read a different block).
can_stored: t.CPtr = t.CPtr(*((uint64*)(block + bs - MEMBUDDY_CANARY)))
can_stored: rawptr = rawptr(*((uint64*)(block + bs - MEMBUDDY_CANARY)))
if uint64(can_stored) != uint64(MEMBUDDY_CANARY_PATTERN):
# Overflow scene: the block-tail canary was overwritten (rc=6). Report the
# block/order/bs plus the index and window start for location.
+1 -1
View File
@@ -6,4 +6,4 @@ extern def va_end(args: t.CPtr)
va_list: t.CTypedef = uchar | t.CPtr
extern def arg(type: t.CType)
extern def arg(type: t.CPtr)
+1 -2
View File
@@ -34,8 +34,7 @@ ULONG: t.CTypedef = t.CUnsignedLong
WCHAR: t.CTypedef = WORD
WCHARPTR: t.CTypedef = WORD | t.CPtr
CHARPTR: t.CTypedef = t.CChar | t.CPtr
FSIZE_t: t.CTypedef = DWORD
LBA_t: t.CTypedef = DWORD
LPCSTR: t.CTypedef = t.CConst | t.CChar | t.CPtr
VOIDPTR: t.CTypedef = t.CPtr
FLOAT: t.CTypedef = t.CFloat