import argparse
import concurrent.futures
import glob
import hashlib
import json
import os
import re
import subprocess
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
NODE = None CACHE_DIR = None OFFLINE = False REPO = "nodejs/node"
def _http_text(url):
req = urllib.request.Request(url, headers={"User-Agent": "jscd-codegen"})
with urllib.request.urlopen(req, timeout=45) as r:
return r.read().decode("utf-8", "replace")
def _cache_path(tag, path):
if not CACHE_DIR:
return None
return os.path.join(CACHE_DIR, tag, path)
def repo_file(tag, path, in_v8=True):
rel = f"deps/v8/{path}" if in_v8 else path
cache = _cache_path(tag, path)
if is_v8_tag(tag):
if not in_v8:
return None
if cache and os.path.exists(cache):
with open(cache, encoding="utf-8") as f:
return f.read()
if cache and os.path.exists(cache + ".404"):
return None
if V8_REPO and os.path.exists(os.path.join(V8_REPO, ".git")):
have = subprocess.run(["git", "-C", V8_REPO, "cat-file", "-e", f"{tag}^{{commit}}"],
capture_output=True, text=True)
if have.returncode == 0:
r = subprocess.run(["git", "-C", V8_REPO, "show", f"{tag}:{path}"],
capture_output=True, text=True)
if r.returncode == 0:
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
with open(cache, "w", encoding="utf-8") as f:
f.write(r.stdout)
return r.stdout
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
open(cache + ".404", "w").close()
return None
return None
if cache and os.path.exists(cache):
with open(cache, encoding="utf-8") as f:
return f.read()
if cache and os.path.exists(cache + ".404"):
return None
git_answered = False
for git_dir in (NODE, REPO_DIR):
if not git_dir or not os.path.exists(os.path.join(git_dir, ".git")):
continue
r = subprocess.run(["git", "-C", git_dir, "cat-file", "-e", f"{tag}^{{commit}}"],
capture_output=True, text=True)
if r.returncode != 0:
continue git_answered = True
r = subprocess.run(["git", "-C", git_dir, "show", f"{tag}:{rel}"],
capture_output=True, text=True)
if r.returncode == 0:
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
with open(cache, "w", encoding="utf-8") as f:
f.write(r.stdout)
return r.stdout
if git_answered:
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
open(cache + ".404", "w").close()
return None
if OFFLINE:
return None
url = f"https://raw.githubusercontent.com/{REPO}/{tag}/{rel}"
try:
text = _http_text(url)
except urllib.error.HTTPError as e:
if e.code == 404:
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
open(cache + ".404", "w").close()
return None
raise
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
with open(cache, "w", encoding="utf-8") as f:
f.write(text)
return text
NEEDED_V8_FILES = [
"include/v8-version.h",
"src/base/functional.h",
"src/base/hashing.h",
"src/snapshot/code-serializer.h",
"src/interpreter/bytecodes.h",
"src/interpreter/bytecode-operands.h",
"src/interpreter/interpreter-intrinsics.h",
"src/snapshot/serializer-deserializer.h",
"src/snapshot/serializer-common.h",
"src/snapshot/serializer.h",
"src/roots/roots.h",
"src/roots.h",
"src/init/heap-symbols.h",
"src/heap-symbols.h",
"src/objects/objects-definitions.h",
"src/objects-definitions.h",
"src/objects/object-list-macros.h",
"src/runtime/runtime.h",
"src/objects/bytecode-array.tq",
"src/objects/code.tq",
"src/objects/trusted-object.tq",
"src/objects/fixed-array.tq",
"src/execution/frame-constants.h",
"src/frame-constants.h",
"src/objects/shared-function-info.tq",
"src/objects/scope-info.tq",
"src/objects/bytecode-array-inl.h",
"src/common/globals.h",
]
REPO_DIR = None REPO_URL = "https://github.com/nodejs/node"
V8_REPO = None
V8_TAGGED_SIZE = 8
V8_SANDBOX = None
V8_PATCHES = [] V8_HASHES = {} V8_TAG_RE = re.compile(r"^(?:\d+\.\d+(?:\.\d+){1,2}|[0-9a-f]{40})$")
def is_v8_tag(tag):
return bool(V8_TAG_RE.match(tag))
def _git(args, **kw):
return subprocess.run(["git", "-C", REPO_DIR, *args], capture_output=True, text=True, **kw)
def ensure_repo():
global REPO_DIR
if REPO_DIR and os.path.exists(os.path.join(REPO_DIR, ".git")):
return
if OFFLINE:
return
d = REPO_DIR or os.path.join(os.path.dirname(__file__), "..", "workspace", "node-partial")
if not os.path.exists(os.path.join(d, ".git")):
os.makedirs(os.path.dirname(d), exist_ok=True)
subprocess.run(["git", "clone", "--filter=blob:none", "--no-checkout", "--depth", "1",
"-q", REPO_URL, d], check=True)
print(f"partial clone -> {d}", file=sys.stderr)
REPO_DIR = d
def have_tag(tag):
r = _git(["rev-parse", "--verify", "--quiet", f"refs/tags/{tag}"])
return r.returncode == 0
def prefetch(tag, jobs=16):
if is_v8_tag(tag):
wanted = [(p, True) for p in NEEDED_V8_FILES]
missing = []
for path, _ in wanted:
cache = _cache_path(tag, path)
if cache and (os.path.exists(cache) or os.path.exists(cache + ".404")):
continue
missing.append((path, path))
if missing and V8_REPO and os.path.exists(os.path.join(V8_REPO, ".git")):
have = subprocess.run(["git", "-C", V8_REPO, "cat-file", "-e", f"{tag}^{{commit}}"],
capture_output=True, text=True)
if have.returncode != 0:
r = subprocess.run(["git", "-C", V8_REPO, "fetch", "--depth", "1",
"--filter=blob:none", "-q", "origin",
f"refs/tags/{tag}:refs/tags/{tag}"],
capture_output=True, text=True)
if r.returncode != 0:
print(f"warn: fetch v8 {tag} failed: {r.stderr.strip()[:120]}",
file=sys.stderr)
_batch_fetch(tag, missing, repo=V8_REPO)
for path, _ in wanted:
repo_file(tag, path, True)
return
wanted = [(p, True) for p in NEEDED_V8_FILES] + [("node.gypi", False), ("common.gypi", False)]
missing = []
for path, in_v8 in wanted:
rel = f"deps/v8/{path}" if in_v8 else path
cache = _cache_path(tag, path)
if cache and (os.path.exists(cache) or os.path.exists(cache + ".404")):
continue
missing.append((path, rel))
ensure_repo()
if REPO_DIR and os.path.exists(os.path.join(REPO_DIR, ".git")):
if not have_tag(tag):
r = _git(["fetch", "--depth", "1", "--filter=blob:none", "-q", "origin",
f"refs/tags/{tag}:refs/tags/{tag}"])
if r.returncode != 0:
print(f"warn: fetch {tag} failed: {r.stderr.strip()[:120]}", file=sys.stderr)
if have_tag(tag) and missing:
_batch_fetch(tag, missing)
for path, in_v8 in wanted:
repo_file(tag, path, in_v8)
def _batch_fetch(tag, missing, repo=None):
req = ("\n".join(f"{tag}:{rel}" for _, rel in missing) + "\n").encode()
r = subprocess.run(["git", "-C", repo or REPO_DIR, "cat-file", "--batch"],
input=req, capture_output=True)
if r.returncode != 0:
return buf = r.stdout
i = 0
for path, _rel in missing:
nl = buf.find(b"\n", i)
if nl < 0:
break
head = buf[i:nl].decode("utf-8", "replace").split()
i = nl + 1
if len(head) == 2 and head[1] == "missing":
cache = _cache_path(tag, path)
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
open(cache + ".404", "w").close()
continue
if len(head) != 3:
break
size = int(head[2])
payload = buf[i:i + size]
i += size + 1 cache = _cache_path(tag, path)
if cache:
os.makedirs(os.path.dirname(cache), exist_ok=True)
with open(cache, "w", encoding="utf-8") as f:
f.write(payload.decode("utf-8", "replace"))
_PATCH_CACHE = {}
def _apply_hunks(text, hunks):
lines = text.splitlines(keepends=True)
cursor = 0
applied = 0
for old, new in hunks:
n = len(old)
found = -1
for i in range(cursor, len(lines) - n + 1):
if lines[i:i + n] == old:
found = i
break
if found < 0:
continue
lines[found:found + n] = new
cursor = found + len(new)
applied += 1
return "".join(lines), applied
def _parse_patch(text):
out = {}
path = None
old, new = [], []
in_hunk = False
def flush():
nonlocal old, new, in_hunk
if path and in_hunk and (old or new):
out.setdefault(path, []).append((old, new))
old, new, in_hunk = [], [], False
for line in text.splitlines(keepends=True):
if line.startswith("diff --git "):
flush()
parts = line.rstrip("\n").split(" b/")
path = parts[-1] if len(parts) > 1 else None
continue
if line.startswith("@@ "):
flush()
in_hunk = True
continue
if not in_hunk:
continue
if line.startswith("\\"):
continue
tag, body = line[:1], line[1:]
if tag == " ":
old.append(body)
new.append(body)
elif tag == "-":
old.append(body)
elif tag == "+":
new.append(body)
flush()
return out
def _patched(tag, path, text):
if not V8_PATCHES:
return text
key = (tag, path)
if key in _PATCH_CACHE:
return _PATCH_CACHE[key]
cur = text
for pf in V8_PATCHES:
try:
hunks_by_path = _parse_patch(open(pf, encoding="utf-8", errors="replace").read())
except OSError as e:
print(f"warn: read patch {pf}: {e}", file=sys.stderr)
continue
hunks = hunks_by_path.get(path)
if not hunks:
continue
cur, applied = _apply_hunks(cur, hunks)
if applied != len(hunks):
print(f"warn: {os.path.basename(pf)}: {path} applied {applied}/{len(hunks)} hunks",
file=sys.stderr)
else:
print(f"patch {os.path.basename(pf)}: {path} ok", file=sys.stderr)
_PATCH_CACHE[key] = cur
return cur
def git_show(tag, path):
text = repo_file(tag, path, in_v8=True)
if text is None:
return None
if V8_PATCHES and is_v8_tag(tag):
return _patched(tag, path, text)
return text
def parse_v8_version(tag):
text = git_show(tag, "include/v8-version.h")
if text is None:
return None
vals = {}
for key in ("V8_MAJOR_VERSION", "V8_MINOR_VERSION", "V8_BUILD_NUMBER", "V8_PATCH_LEVEL"):
m = re.search(rf"#define {key} (\d+)", text)
if not m:
return None
vals[key] = int(m.group(1))
return (vals["V8_MAJOR_VERSION"], vals["V8_MINOR_VERSION"],
vals["V8_BUILD_NUMBER"], vals["V8_PATCH_LEVEL"])
def extract_bytecodes(text):
m = re.search(r"#define BYTECODE_LIST(?:\w*)\(([^)]*)\)\s*\\(.*?)(?=\n#define |\n#endif|\Z)",
text, re.S)
if not m:
raise RuntimeError("BYTECODE_LIST not found")
body = m.group(2)
entries = []
for line in body.splitlines():
line = line.strip()
lm = re.match(r"^V\(\s*(\w+)\s*,", line)
if not lm:
continue
name = lm.group(1)
operands = re.findall(r"OperandType::k(\w+)", line)
is_prefix = name in ("Wide", "ExtraWide")
entries.append({"name": name, "operands": operands, "prefix": is_prefix or None})
return entries
def extract_operand_types(text):
INFO = {
"None": (0, False),
"ScalableSignedByte": (1, True),
"ScalableUnsignedByte": (1, True),
"FixedSignedByte": (1, False),
"FixedUnsignedByte": (1, False),
"FixedUnsignedShort": (2, False),
"FixedUnsignedQuad": (4, False),
"FixedSignedShort": (2, False),
}
types = {}
for m in re.finditer(r"V\(\s*(\w+)\s*,\s*OperandTypeInfo::k(\w+)\s*\)", text):
name, info = m.group(1), m.group(2)
if info in INFO:
size, scalable = INFO[info]
types[name] = {"size": size, "scalable": scalable}
for m in re.finditer(r"V\(\s*(\w+)\s*,\s*OperandSize::k(\w+)\s*(?:,\s*OperandScale::k(\w+))?\s*\)", text):
name, size, scale = m.group(1), m.group(2), m.group(3)
size_bytes = {"None": 0, "Byte": 1, "Short": 2, "Quad": 4}.get(size)
if size_bytes is None:
continue
scalable = scale in ("Double", "Quadruple", "Octuple")
types[name] = {"size": size_bytes, "scalable": scalable}
return types
def extract_legacy_tags(text):
if text is None:
return None
sect = text[text.find("class SerializerDeserializer") :]
consts = dict(
(k, int(v, 0)) for k, v in re.findall(r"static const int (k\w+)\s*=\s*(0x[0-9a-fA-F]+|\d+)\s*;", sect)
)
for m in re.finditer(r"enum (Where|HowToCode|WhereToPoint)\s*\{(.*?)\}", sect, re.S):
for k, v in re.findall(r"(k\w+)\s*=\s*(0x[0-9a-fA-F]+|\d+)", m.group(2)):
consts.setdefault(k, int(v, 0))
return consts or None
def extract_serialization_tags(text):
m = re.search(r"enum Bytecode(?:\s*:\s*(?:byte|uint8_t|uint8))?\s*\{(.*?)\n \};", text, re.S)
if not m:
raise RuntimeError("SerializationTag::Bytecode not found")
body = m.group(1)
tags = {}
value = 0
for raw_line in body.splitlines():
line = raw_line.split("//")[0].strip()
line = line.rstrip(",").rstrip("\\").strip()
if not line or line.startswith("/*") or line.startswith("*"):
continue
dm = re.match(r"(\w+)\s*=\s*(0x[0-9a-fA-F]+|\d+)\s*$", line)
if dm:
tags[dm.group(1)] = int(dm.group(2), 0)
value = tags[dm.group(1)] + 1
continue
if re.match(r"^\w+$", line):
tags[line] = value
value += 1
return tags
def _grab_macro_list(text, name):
m = re.search(rf"#define {name}\(([^)]*)\)(.*?)(?=\n#define |\Z)", text, re.S)
return m.group(2) if m else None
def _acc_use(inner):
uses = set(re.findall(r"(?:ImplicitRegisterUse|AccumulatorUse)::(k\w+)", inner))
if not uses:
return ""
read = bool(uses & {"kRead", "kReadWrite",
"kReadAccumulator", "kReadWriteAccumulator",
"kReadAccumulatorWriteShortStar", "kWriteShortStar"})
write = bool(uses & {"kWrite", "kReadWrite",
"kWriteAccumulator", "kReadWriteAccumulator"})
return ("rw" if (read and write) else "r" if read else "w" if write else "")
def _expand_bytecode_entries(text, macro_name, entries, seen=None, depth=0):
if seen is None:
seen = set()
if macro_name in seen or depth > 8:
return
seen.add(macro_name)
body = _grab_macro_list(text, macro_name)
if body is None:
return
i = 0
while i < len(body):
im = re.compile(r"\b(\w+)\s*\(").search(body, i)
if not im:
break
ident = im.group(1)
j = body.index("(", im.start())
depth_p = 0
k = j
while k < len(body):
if body[k] == "(":
depth_p += 1
elif body[k] == ")":
depth_p -= 1
if depth_p == 0:
break
k += 1
inner = body[j + 1:k]
i = k + 1
if ident in ("V", "V_TSA"):
args = [a.strip() for a in inner.split(",")]
name = args[0]
m2 = re.match(r"^(\w+)$", name)
if not m2 or name in ("ImplicitRegisterUse", "OperandScale"):
continue
operands = re.findall(r"OperandType::k(\w+)", inner)
is_prefix = name in ("Wide", "ExtraWide")
entries.append({
"name": name,
"operands": operands,
"prefix": name if is_prefix else None,
"acc": _acc_use(inner),
})
else:
_expand_bytecode_entries(text, ident, entries, seen, depth + 1)
def extract_bytecodes(text):
if not re.search(r"#define BYTECODE_LIST\(", text):
raise RuntimeError("BYTECODE_LIST not found")
entries = []
_expand_bytecode_entries(text, "BYTECODE_LIST", entries)
return entries
def _split_top_args(inner):
args, depth, cur, in_str = [], 0, "", None
for ch in inner:
if in_str:
cur += ch
if ch == in_str:
in_str = None
continue
if ch in ("\"", "'"):
in_str = ch
cur += ch
continue
if ch == "(":
depth += 1
elif ch == ")":
depth -= 1
if ch == "," and depth == 0:
args.append(cur.strip())
cur = ""
else:
cur += ch
if cur.strip():
args.append(cur.strip())
return args
def _roots_six_x(*sources):
texts = [t for t in sources if t]
def body(name):
for t in texts:
m = re.search(rf"#define {name}\(V\)(.*?)(?=\n#define |\Z)", t, re.S)
if m:
return m.group(1)
return None
def entries(name):
b = body(name)
if b is None:
return None
b = b.replace("\\\n", " ")
out = []
for m in re.finditer(r"\bV\(([^()]*)\)", b):
out.append(_split_top_args(m.group(1)))
return out
out = []
for a in entries("STRONG_ROOT_LIST") or []:
out.append(a[2] if len(a) >= 3 else a[0])
for a in entries("INTERNALIZED_STRING_LIST") or []:
lit = a[1].strip()
if lit.startswith('"') and lit.endswith('"'):
lit = lit[1:-1]
out.append("String:" + lit)
for name in ("PRIVATE_SYMBOL_LIST", "PUBLIC_SYMBOL_LIST", "WELL_KNOWN_SYMBOL_LIST"):
for a in entries(name) or []:
out.append("Symbol:" + a[0])
for a in entries("ACCESSOR_INFO_LIST") or []:
out.append(a[1] if len(a) >= 2 else a[0])
for a in entries("STRUCT_LIST") or []:
out.append(a[1] if len(a) >= 2 else a[0])
for a in entries("DATA_HANDLER_LIST") or []:
out.append((a[1] + a[2] + "Map") if len(a) >= 3 else a[0])
out.append("StringTable")
for a in entries("SMI_ROOT_LIST") or []:
out.append(a[2] if len(a) >= 3 else a[0])
return out
def c_string_literal(arg):
s = arg.strip()
if len(s) >= 2 and s[0] == '"' and s[-1] == '"':
s = s[1:-1]
out, i = [], 0
while i < len(s):
if s[i] == "\\" and i + 1 < len(s) and s[i + 1] in ('"', "'", "\\"):
out.append(s[i + 1])
i += 2
else:
out.append(s[i])
i += 1
return "".join(out)
def extract_roots(text, symbols_text=None, defs_text=None, torque_count=None):
if torque_count is None:
torque_count = TORQUE_MAP_COUNT
if text and "STRONG_ROOT_LIST" in text:
return _roots_six_x(text, symbols_text, defs_text)
sources = [text] + [s for s in (symbols_text, defs_text) if s]
def grab(name):
for s in sources:
m = re.search(rf"#define {name}\b(.*?)(?=\n#define |\Z)", s, re.S)
if m:
return m.group(1)
return None
order = []
def split_top(inner):
args, depth, cur, in_str, esc = [], 0, "", None, False
for ch in inner:
if in_str:
cur += ch
if esc:
esc = False
elif ch == "\\":
esc = True
elif ch == in_str:
in_str = None
continue
if ch in ("\"", "'"):
in_str = ch
cur += ch
continue
if ch == "(":
depth += 1
elif ch == ")":
depth -= 1
if ch == "," and depth == 0:
args.append(cur.strip())
cur = ""
else:
cur += ch
if cur.strip():
args.append(cur.strip())
return args
def invocations(body):
lines = []
for raw in body.splitlines():
code = raw.split("//")[0]
lines.append(code)
body = "\n".join(lines).replace("\\\n", " ") i = 0
while i < len(body):
m = re.compile(r"\b(\w+)\s*\(").search(body, i)
if not m:
break
ident = m.group(1)
j = body.index("(", m.start())
d, k, in_str = 0, j, None
while k < len(body):
ch = body[k]
if in_str:
if ch == "\\":
k += 2
continue
if ch == in_str:
in_str = None
elif ch in ("\"", "'"):
in_str = ch
elif ch == "(":
d += 1
elif ch == ")":
d -= 1
if d == 0:
break
k += 1
yield ident, body[j + 1:k]
i = k + 1
def add_generator_entry(gen, inner, mode):
args = split_top(inner)
if len(args) < 2:
return
if "INTERNALIZED_STRING" in gen:
if len(args) >= 3 and args[2].startswith('"'):
literal = c_string_literal(args[2])
order.append("String:" + literal)
else:
order.append("String:" + args[1].removesuffix("_string"))
elif "SYMBOL" in gen:
order.append("Symbol:" + args[1])
elif "ACCESSOR_INFO" in gen:
order.append("AccessorInfo:" + args[1] + "_accessor")
elif gen.startswith("STRUCT_LIST_GENERATOR"):
order.append((args[2] if len(args) >= 3 else args[1]) + "Map")
elif mode == "maps" and gen in ("ALLOCATION_SITE_LIST", "DATA_HANDLER_LIST"):
order.append((args[1] + args[2] + "Map") if len(args) >= 3 else args[1] + "Map")
else:
order.append("Gen:" + args[1])
def expand_macro(name, depth=0, mode=None):
if depth > 10:
return
if name == "TORQUE_DEFINED_MAP_ROOT_LIST":
order.append("__torque_map_list_unresolved__")
return
body = grab(name)
if body is None:
return
is_generator = "_GENERATOR" in name
my_mode = "maps" if name.endswith("_MAPS_LIST") else mode
for ident, inner in invocations(body):
if ident in ("V", "V_"):
if is_generator:
add_generator_entry(name, inner, my_mode)
else:
args = split_top(inner)
if len(args) >= 3:
if args[0].strip() == "String" and args[2] == "empty_string":
order.append("String:")
else:
order.append(args[2])
elif len(args) == 2:
order.append(args[1])
elif ident == "IF_WASM":
args = split_top(inner)
if my_mode == "maps" and len(args) >= 4:
order.append(args[3] + "Map")
elif len(args) >= 4:
order.append(args[3])
else:
expand_macro(ident, depth + 1, my_mode)
for top in ("READ_ONLY_ROOT_LIST", "MUTABLE_ROOT_LIST"):
expand_macro(top)
n = torque_count
if "__torque_map_list_unresolved__" in order:
cutoff = order.index("__torque_map_list_unresolved__")
head, tail = order[:cutoff], order[cutoff + 1 :]
order = head + [f"__torque_map_{i}__" for i in range(n)] + tail
return order
TABLE_ALIAS: dict[str, str] = {}
TORQUE_MAP_COUNT_BY_VERSION = {
"10_6": 25, "11_0": 34, "11_2": 34, "11_4": 34, "11_6": 34, "12_0": 33, "12_2": 33, "11_3": 34,
"10_7": 25, "10_8": 25, "11_8": 34, "14_1": 32, "14_0": 32, "14_2": 32, "14_4": 32, "14_6": 32, "14_8": 32, }
TORQUE_MAP_COUNT = 36
def extract_header_layout(tag):
text = git_show(tag, "src/snapshot/code-serializer.h")
if text is None:
return None
consts = {"kMagicNumberOffset": 0, "kUInt32Size": 4}
for m in re.finditer(r"static const uint32_t (k\w+)\s*=\s*([^;]+);", text):
name, expr = m.group(1), m.group(2).strip()
expr = re.sub(r"POINTER_SIZE_ALIGN\(([^)]*)\)", r"ALIGN8(\1)", expr)
try:
val = _eval_expr(expr, consts)
except Exception:
continue
consts[name] = val
layout = {
"version_hash": consts.get("kVersionHashOffset", 4),
"source_hash": consts.get("kSourceHashOffset", 8),
"flag_hash": consts.get("kFlagHashOffset", 12),
"read_only_checksum": consts.get("kReadOnlySnapshotChecksumOffset"),
"payload_length": consts.get("kPayloadLengthOffset", 16),
"checksum": consts.get("kChecksumOffset", 20),
"header_size": consts.get("kHeaderSize", 24),
}
return layout
def _eval_expr(expr, consts):
expr = expr.strip()
expr = expr.replace("POINTER_SIZE_ALIGN", "ALIGN8")
expr = re.sub(r"k\w+", lambda m: str(consts.get(m.group(0), 0)), expr)
m = re.match(r"^ALIGN8\((.*)\)$", expr, re.S)
if m:
inner = _eval_expr(m.group(1), consts)
return (inner + 7) & ~7
if not re.fullmatch(r"[0-9+\-*/()\s~&|]+", expr):
raise ValueError(f"unsupported expr: {expr!r}")
return int(eval(expr))
BUILD_DEFINES = {
"V8_TRACE_UNOPTIMIZED": False, "V8_TRACE_FEEDBACK_UPDATES": False, "V8_INTL_SUPPORT": True, "V8_ENABLE_WEBASSEMBLY": True, "V8_TRACE_IGNITION": False, "V8_DUMPLING": False,
}
IF_FLAGS = {
"IF_WASM": True, "IF_WASM_DRUMBRAKE": False,
"IF_SPARKPLUG_PLUS": True,
"IF_V8_WASM_RANDOM_FUZZERS": False,
"IF_TSA": True,
}
def strip_disabled_blocks(text):
out, stack = [], [] for line in text.splitlines():
st = line.strip()
enabled = stack[-1][0] if stack else True
if st.startswith("#ifdef ") or st.startswith("#ifndef "):
parts = st.split()
flag = parts[1] if len(parts) > 1 else ""
if flag not in BUILD_DEFINES:
on = True
else:
on = BUILD_DEFINES[flag] if st.startswith("#ifdef ") else not BUILD_DEFINES[flag]
stack.append([enabled and on, enabled])
continue
if st.startswith("#else") and stack:
top = stack[-1]
top[0] = top[1] and not top[0]
continue
if st.startswith("#endif") and stack:
stack.pop()
continue
if enabled:
out.append(line)
return "\n".join(out)
def extract_runtime_names(text):
if text is None:
return []
text = strip_disabled_blocks(text)
def macro_body(name):
m = re.search(rf"#define {name}\(F, I\)(.*?)(?=\n#define |\Z)", text, re.S)
if m:
return m.group(1)
m = re.search(rf"#define {name}\(F\)(.*?)(?=\n#define |\Z)", text, re.S)
return m.group(1) if m else None
def invocations(body):
body = body.replace("\\\n", " ")
i = 0
while i < len(body):
m = re.compile(r"\b([A-Za-z_][A-Za-z0-9_]*)\s*\(").search(body, i)
if not m:
break
ident = m.group(1)
j = body.index("(", m.start())
d, k = 0, j
while k < len(body):
if body[k] == "(":
d += 1
elif body[k] == ")":
d -= 1
if d == 0:
break
k += 1
yield ident, body[j + 1:k]
i = k + 1
out = []
inline = []
split_fi = not bool(re.search(r"#define FOR_EACH_INTRINSIC_RETURN_PAIR_IMPL\b", text))
def expand(macro_name, depth=0):
if depth > 6:
return
for flag, enabled in BUILD_DEFINES.items():
if macro_name.startswith(f"FOR_EACH_INTRINSIC_{flag.removeprefix('V8_').split('_')[0]}"):
pass
if macro_name == "FOR_EACH_INTRINSIC_TRACE_UNOPTIMIZED" and not BUILD_DEFINES["V8_TRACE_UNOPTIMIZED"]:
return
if macro_name == "FOR_EACH_INTRINSIC_TRACE_FEEDBACK" and not BUILD_DEFINES["V8_TRACE_FEEDBACK_UPDATES"]:
return
if macro_name == "FOR_EACH_INTRINSIC_INTL" and not BUILD_DEFINES["V8_INTL_SUPPORT"]:
return
if macro_name == "FOR_EACH_INTRINSIC_WASM" and not BUILD_DEFINES["V8_ENABLE_WEBASSEMBLY"]:
return
body = macro_body(macro_name)
if body is None:
return
for ident, inner in invocations(body):
if ident in ("F", "I"):
name = inner.split(",")[0].strip()
if name and name[0].isalpha():
if split_fi:
(out if ident == "F" else inline).append(name)
else:
out.append(name)
elif ident.startswith("IF_"):
enabled = IF_FLAGS.get(ident, True)
if enabled:
parts = [p.strip() for p in inner.split(",")]
if parts and parts[0].startswith("FOR_EACH_"):
expand(parts[0], depth + 1)
elif parts and parts[0] in ("F", "I"):
name = parts[1].strip() if len(parts) > 1 else ""
if name and name[0].isalpha():
out.append(name)
elif ident.startswith("FOR_EACH_"):
expand(ident, depth + 1)
expand("FOR_EACH_INTRINSIC_RETURN_PAIR_IMPL")
expand("FOR_EACH_INTRINSIC_RETURN_OBJECT_IMPL")
if not out:
expand("FOR_EACH_INTRINSIC")
return out + (inline if inline else list(out))
return out + inline
def extract_intrinsic_names(text):
if text is None:
return []
m = re.search(r"#define INTRINSICS_LIST\(V\)(.*?)(?=\n#define |\Z)", text, re.S)
if not m:
return []
body = m.group(1).replace("\\\n", " ")
return re.findall(r"V\(\s*([A-Za-z0-9_]+)\s*,", body)
def _tq_class(text, name):
m = re.search(rf"(?:extern )?class {name}\b([^{{]*)\{{", text)
if not m:
return None, None
header = m.group(1)
i, depth, start = m.end(), 1, m.end()
while i < len(text) and depth:
c = text[i]
if c == "{":
depth += 1
elif c == "}":
depth -= 1
i += 1
body = text[start:i - 1] if depth == 0 else text[start:]
bm = re.search(r"extends\s+(\w+)", header)
return body, (bm.group(1) if bm else None)
def _parse_field_line(line):
line = re.sub(r"^(weak|const|extern|static)\s+", "", line.strip())
m = re.match(r"^const\s+([a-z_][a-z0-9_]*)\s*:\s*(.+?);?$", line)
if not m:
m = re.match(r"^([a-z_][a-z0-9_]*)\s*(\[[^\]]*\])?\s*:\s*(.+?);?$", line)
if not m:
return None
name, bracket, ty = m.group(1), m.group(2), m.group(3)
return name, ty, bool(bracket)
return m.group(1), m.group(2), False
def _tq_type_size(ty, sources, tagged_size, sandbox, depth=0):
ty = ty.strip()
if ty in ("void", "constexpr '')"):
return 0
if "|" in ty:
return tagged_size
base = re.split(r"[<:]", ty)[0].strip()
simple = {"int32": 4, "uint32": 4, "int16": 2, "uint16": 2,
"int8": 1, "uint8": 1, "bool": 1, "Smi": tagged_size}
if base in simple:
return simple[base]
if base in ("ProtectedPointer", "IndirectPointer", "ProtectedPointerTagged"):
return 4 if sandbox else tagged_size
if base == "Indirect": return 4 if sandbox else tagged_size
for text in sources.values():
bm = re.search(rf"bitfield struct {base}\s+extends\s+(\w+)", text)
if bm:
return _tq_type_size(bm.group(1), sources, tagged_size, sandbox, depth + 1)
if depth < 4:
for text in sources.values():
body, base_name = _tq_class(text, base)
if body is None:
continue
if not _is_inline_struct(sources, base_name):
break total = 0
for line, cond, _ in _tq_fields(body, tagged_size, sandbox):
if cond and not _cond_enabled(cond, tagged_size, sandbox):
continue
parsed = _parse_field_line(line)
if parsed and not re.fullmatch(r"padding\d*", parsed[0]):
total += _tq_type_size(parsed[1], sources, tagged_size, sandbox, depth + 1)
return total
return tagged_size
def _is_inline_struct(sources, class_name, depth=0):
if not class_name or depth > 4:
return False
if class_name == "Struct":
return True
for text in sources.values():
body, base = _tq_class(text, class_name)
if body is not None:
return _is_inline_struct(sources, base, depth + 1)
return False
def _cond_enabled(cond, tagged_size, sandbox):
cond = cond.strip()
negate = False
if cond.startswith("ifnot(") or cond.startswith("ifnot ("):
negate = True
cond = cond[cond.index("(") + 1 : cond.rindex(")")]
elif cond.startswith("if(") or cond.startswith("if ("):
cond = cond[cond.index("(") + 1 : cond.rindex(")")]
flags = {
"TAGGED_SIZE_8_BYTES": tagged_size == 8,
"TAGGED_SIZE_4_BYTES": tagged_size == 4,
"V8_ENABLE_SANDBOX": sandbox,
"V8_EXTERNAL_CODE_SPACE": False,
"V8_ENABLE_LEAPTIERING": True,
"V8_ENABLE_WEBASSEMBLY": BUILD_DEFINES["V8_ENABLE_WEBASSEMBLY"],
"V8_INTL_SUPPORT": BUILD_DEFINES["V8_INTL_SUPPORT"],
}
val = flags.get(cond, True)
return (not val) if negate else val
def _tq_fields(body, tagged_size, sandbox):
out = []
pending_cond = None
for raw in body.splitlines():
line = raw.split("//")[0].strip()
if not line:
continue
cond = pending_cond
pending_cond = None
while line.startswith("@"):
cm = re.match(r"@(if|ifnot)\s*\(([^)]*)\)\s*(.*)$", line)
if cm:
c = f"{cm.group(1)}({cm.group(2)})"
rest = cm.group(3).strip()
if rest:
out.append((rest, c, None))
line = ""
else:
pending_cond = c
line = ""
break
am = re.match(r"@\w+\s*(.*)$", line)
rest = am.group(1).strip() if am else ""
line = rest
if not line:
break
if line and not line.startswith("@"):
out.append((line, cond, None))
return out
def tq_layout(sources, class_name, tagged_size, sandbox):
chain = []
cur = class_name
guard = 0
while cur and guard < 8:
guard += 1
found = False
for text in sources.values():
body, base = _tq_class(text, cur)
if body is not None:
chain.append((cur, body))
cur = base
found = True
break
if not found:
break
if not chain:
return None
offset = tagged_size fields = {}
for name, body in reversed(chain):
for fname, cond, _ in _tq_fields(body, tagged_size, sandbox):
if cond and not _cond_enabled(cond, tagged_size, sandbox):
continue
parsed = _parse_field_line(fname)
if not parsed:
continue
field, ty, is_var = parsed
if re.fullmatch(r"padding\d*", field):
continue
if is_var:
fields.setdefault(f"__var_{field}", offset)
continue
size = _tq_type_size(ty, sources, tagged_size, sandbox)
fields[field] = offset
offset += size
return {"header_size": offset, "fields": fields}
def bytecode_array_layout(sources, tagged_size, sandbox):
return tq_layout(sources, "BytecodeArray", tagged_size, sandbox)
FRAME_OVERRIDES = {
"8.7.220.31": {
"reg_file_start": -6,
"context_index": -5,
"closure_index": -4,
"first_param": -8,
"drop_param_base": True,
},
}
def apply_frame_override(tag, frame):
ov = FRAME_OVERRIDES.get(tag)
if not ov or frame is None:
return frame
frame = dict(frame)
for k in ("reg_file_start", "context_index", "closure_index", "first_param"):
if k in ov:
frame[k] = ov[k]
if ov.get("drop_param_base"):
frame.pop("param_base", None)
return frame
def extract_frame_layout(text):
if text is None:
return None
m = re.search(r"class UnoptimizedFrameConstants.*?DEFINE_STANDARD_FRAME_SIZES\((\d+)\)", text, re.S)
if not m:
return extract_frame_layout_legacy(text)
extra = int(m.group(1))
cp_slots = 0
standard = 3 + cp_slots fixed = standard + extra
start = -(fixed + 1) return {
"reg_file_start": start,
"context_index": start + 1,
"closure_index": start + 2,
"first_param": start - 2,
"extra_slots": extra,
"cp_slots": cp_slots,
}
def extract_frame_layout_legacy(text):
if text is None or "kRegisterFileFromFp" not in text:
return None
cp_slots = 0 standard = 2 + cp_slots
start = -(standard + 3)
return {
"reg_file_start": start,
"context_index": start + 1,
"closure_index": start + 2 + cp_slots,
"last_param": start - 2,
"param_base": start - 1,
"extra_slots": 2,
"cp_slots": cp_slots,
}
def extract_parameter_count_semantics(text):
if text is None:
return {"direct": False}
m = re.search(r"parameter_count\(\) const \{(.*?)\n\}", text, re.S)
if m and re.search(r"ReadField<uint16_t>", m.group(1)):
return {"direct": True}
if re.search(r"uint16_t\s+parameter_size_", text):
return {"direct": True}
return {"direct": False}
def extract_scope_info(text, globals_text):
if text is None:
return None
max_inlined = 75
if globals_text:
m = re.search(r"kScopeInfoMaxInlinedLocalNamesSize\s*=\s*(\d+)", globals_text)
if m:
max_inlined = int(m.group(1))
flags_smi = "SmiTagged<ScopeFlags>" in text
has_table = "context_local_names_hashtable" in text
m_pos = text.find("position_info")
m_names = text.find("context_local_names")
early_position = m_pos != -1 and m_names != -1 and m_pos < m_names
out = {
"flags_smi": flags_smi,
"position_info_early": early_position,
"max_inlined_names": max_inlined if has_table else (1 << 30),
"saved_class_bit": 10, "function_variable_bits": [12, 13],
"receiver_bits": [7, 8],
"has_inferred_bit": 14,
}
bits = scope_flags_bits_from_chain(text)
if bits:
out.update(bits)
return out
def scope_flags_bits_from_chain(text):
flat = re.sub(r"\s+", " ", text)
nxt = re.compile(r"using (\w+?)Field = (\w+?)Field::Next<([\w:]+), (\d+)>;")
head = re.compile(r"using (\w+?)Field = BitField<([\w:]+), (\d+), (\d+)>;")
succ = {prev: (name, int(width)) for name, prev, _ty, width in nxt.findall(flat)}
heads = head.findall(flat)
if not heads:
return None
order, cur, bit = [], heads[0][0], int(heads[0][2])
seen = set()
while cur and cur not in seen:
seen.add(cur)
if cur in succ:
name, width = succ[cur]
order.append((cur, bit, width))
bit += width
cur = name
else:
break
starts = {name: (start, width) for name, start, width in order}
if "ReceiverVariable" not in starts or "FunctionVariable" not in starts:
return None
rs, rw = starts["ReceiverVariable"]
fs, fw = starts["FunctionVariable"]
out = {
"receiver_bits": [rs, rs + rw - 1],
"function_variable_bits": [fs, fs + fw - 1],
}
if "HasInferredFunctionName" in starts:
out["has_inferred_bit"] = starts["HasInferredFunctionName"][0]
if "HasSavedClassVariableIndex" in starts:
out["saved_class_bit"] = starts["HasSavedClassVariableIndex"][0]
else:
out["saved_class_bit"] = 30 return out
def extract_hash_fold(tag):
for path in ("src/base/functional.h", "src/base/hashing.h"):
fh = git_show(tag, path)
if fh and "class Hasher" in fh and "hash_value_unsigned_impl" in fh:
return "left_fold"
vv = parse_v8_version(tag)
if vv and (vv[0] > 12 or (vv[0] == 12 and vv[1] >= 2)):
return "left_fold"
return "right_fold"
def sandbox_of(tag):
if V8_SANDBOX is not None:
return V8_SANDBOX
return extract_tagged_size(tag) == 4
def extract_tagged_size(tag):
if is_v8_tag(tag):
return V8_TAGGED_SIZE
gypi = repo_file(tag, "node.gypi", in_v8=False) or ""
gypi += repo_file(tag, "common.gypi", in_v8=False) or ""
enabled = re.search(r"'v8_enable_pointer_compression':\s*(\d)", gypi)
return 4 if (enabled and enabled.group(1) == "1") else 8
MAGIC_BASE = 0xC0DE0000
def v8_hash(fold, a, b, c, d):
M = 0xC6A4A7935BD1E995
MASK = (1 << 64) - 1
def mix32(v):
v &= 0xFFFFFFFF
v = (~v & 0xFFFFFFFF) + ((v << 15) & 0xFFFFFFFF)
v &= 0xFFFFFFFF
v ^= v >> 12
v = (v + ((v << 2) & 0xFFFFFFFF)) & 0xFFFFFFFF
v ^= v >> 4
v = (v * 2057) & 0xFFFFFFFF
v ^= v >> 16
return v
def combine(seed, val):
val = (val * M) & MASK
val ^= val >> 47
val = (val * M) & MASK
seed ^= val
seed = (seed * M) & MASK
return seed
s = 0
order = [d, c, b, a] if fold == "right_fold" else [a, b, c, d]
for v in order:
s = combine(s, mix32(v))
return s & 0xFFFFFFFF
def _deep_fill(dst, src, path=""):
if not isinstance(dst, dict) or not isinstance(src, dict):
return 0
n = 0
for k, v in src.items():
if k not in dst or dst[k] is None:
dst[k] = v
n += 1
elif isinstance(dst[k], dict) and isinstance(v, dict):
n += _deep_fill(dst[k], v, f"{path}{k}.")
return n
def donor_for(minor, donors):
if not donors:
return None
def key(m):
a, b = (int(x) for x in m.split("."))
return a * 100 + b
tgt = key(minor)
major = int(minor.split(".")[0])
pool = list(donors)
pool = list(donors)
if False:
pass
return min(pool, key=lambda m: abs(key(m) - tgt))
OPERAND_TYPE_ALIASES = {
"ConstantPoolIndex": "Idx", "ContextSlot": "Idx", "FeedbackSlot": "Idx", "CoverageSlot": "Idx", "AbortReason": "Flag8", "EmbeddedFeedback": "Flag16",
}
def normalize_operand_types(table):
aliases = dict(OPERAND_TYPE_ALIASES)
emb = table.get("operand_types", {}).get("EmbeddedFeedback")
if emb is not None:
aliases["EmbeddedFeedback"] = {1: "Flag8", 2: "Flag16"}.get(
int(emb.get("size") or 0), "Flag16"
)
table["bytecodes"] = [
{**b, "operands": [aliases.get(o, o) for o in b.get("operands", [])]}
for b in table["bytecodes"]
]
merged = {}
for k, v in table["operand_types"].items():
merged.setdefault(aliases.get(k, k), v)
table["operand_types"] = merged
return table
def ver_key(version):
try:
return tuple(int(x) for x in version.lstrip("v").split("."))
except ValueError:
return (10 ** 9,)
def load_index(path, floor=None):
if not os.path.exists(path):
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
json.dump(json.loads(_http_text("https://nodejs.org/dist/index.json")), f)
entries = json.load(open(path, encoding="utf-8"))
if floor:
entries = [e for e in entries if ver_key(e["version"]) >= floor]
return entries
def tags_for_all(entries, major_floor=None):
newest = {}
for e in entries:
v8 = e.get("v8")
if not v8:
continue
if major_floor and ver_key(e["version"]) < major_floor:
continue
cur = newest.get(v8)
if cur is None or ver_key(e["version"]) > ver_key(cur):
newest[v8] = e["version"]
return sorted(set(newest.values()), key=ver_key)
def main():
global NODE, CACHE_DIR, OFFLINE
ap = argparse.ArgumentParser()
ap.add_argument("--node", default=None,
help="本地 node 克隆（可选；给了就优先用它，离线也能跑）")
ap.add_argument("--tag", action="append", default=[],
help="node tag，可多次；不给则用 --all-from")
ap.add_argument("--all-from", default=None,
help="从 dist/index.json 取所有 >= 该版本的发布（如 8.0.0），"
"每个不同的 V8 版本挑一个 tag")
ap.add_argument("--index", default=None,
help="nodejs.org/dist/index.json 缓存（缺省 workspace/node-dist-index.json，"
"没有就自动下载一次）")
ap.add_argument("--cache", default=None,
help="单文件源码缓存目录（缺省 workspace/node-src）")
ap.add_argument("--offline", action="store_true", help="只用缓存/克隆，不联网")
ap.add_argument("--repo-dir", default=None,
help="blob 过滤的部分克隆目录（缺省 workspace/node-partial，自动创建）")
ap.add_argument("--no-clone", action="store_true",
help="不建部分克隆，直接走 raw.githubusercontent")
ap.add_argument("--jobs", type=int, default=16, help="并发取文件数（默认 16）")
ap.add_argument("--donors", default=None,
help="已知表的目录（tables/）：新表缺的字段从最近的 donor 借")
ap.add_argument("--keep-existing", action="store_true",
help="输出目录里已存在的 v<major>_<minor>.json 直接沿用，不覆盖")
ap.add_argument("--ro-maps", default=None,
help="已生成的 ro-map 目录（workspace/ro）：按 V8 minor 收进 tables/ 并内嵌")
ap.add_argument("--per-minor", action="store_true",
help="每个 V8 major.minor 只保留最新 patch 的表（其余发布映射到它）")
ap.add_argument("--v8-repo", default=None,
help="v8/v8 的部分克隆（blob 过滤 + depth 1）；配合 --v8-tag 取 Node 未发布的 V8 版本")
ap.add_argument("--v8-tag", action="append", default=[],
help="V8 版本号（如 15.6.75.19），可多次；用于 Electron/Chrome 独有版本")
ap.add_argument("--v8-tagged-size", type=int, default=8,
help="--v8-tag 构建的 kTaggedSize（Chromium/Electron 是 4 = 指针压缩）")
ap.add_argument("--v8-hash", action="append", default=[],
help="TAG=0x...：用真缓存里读到的 version_hash 覆盖计算值（Electron 加盐）")
ap.add_argument("--v8-patches", default=None,
help="Electron 的 V8 补丁目录（*.patch）：取数时把命中文件的 hunk 打上")
ap.add_argument("--v8-sandbox", choices=["auto", "on", "off"], default="auto",
help="--v8-tag 构建是否开 V8 sandbox（auto：ts==4 即视为开）")
ap.add_argument("--v8-label", action="append", default=[],
help="V8_TAG=标签，写进 manifest 的 node 字段（如 15.6.75.19=electron 45.0.0）")
ap.add_argument("--index-entries", action="store_true",
help="只按 --index 给**已有表**补 manifest 发布条目（不抓源码、不重建表）")
ap.add_argument("--merge-existing", action="store_true",
help="在现有 tables/manifest.json 上追加：保留既有条目/表/ro-map，只生成新 tag 的表")
ap.add_argument("--out", default=None, help="输出目录（默认仓库根）")
args = ap.parse_args()
NODE = args.node
OFFLINE = args.offline
global V8_REPO, V8_TAGGED_SIZE, V8_SANDBOX, V8_PATCHES, V8_HASHES
V8_REPO = args.v8_repo
if args.v8_patches:
V8_PATCHES = sorted(glob.glob(os.path.join(args.v8_patches, "*.patch")))
for item in args.v8_hash:
if "=" in item:
k, v = item.split("=", 1)
try:
V8_HASHES[k.strip()] = int(v.strip(), 0)
except ValueError:
ap.error(f"bad --v8-hash value: {item}")
V8_TAGGED_SIZE = args.v8_tagged_size
V8_SANDBOX = None if args.v8_sandbox == "auto" else (args.v8_sandbox == "on")
v8_labels = {}
for item in args.v8_label:
if "=" in item:
k, v = item.split("=", 1)
v8_labels[k.strip()] = v.strip()
out_root = args.out or os.path.join(os.path.dirname(__file__), "..")
CACHE_DIR = args.cache or os.path.join(out_root, "workspace", "node-src")
global REPO_DIR
if not args.no_clone:
REPO_DIR = args.repo_dir or os.path.join(out_root, "workspace", "node-partial")
tables_dir = os.path.join(out_root, "tables")
os.makedirs(tables_dir, exist_ok=True)
merged_versions, merged_tables, merged_ro = [], {}, {}
if args.merge_existing:
mpath = os.path.join(tables_dir, "manifest.json")
if os.path.exists(mpath):
with open(mpath, encoding="utf-8") as f:
merged_versions = json.load(f).get("versions", [])
for f in sorted(glob.glob(os.path.join(tables_dir, "v*.json"))):
try:
with open(f, encoding="utf-8") as fh:
merged_tables[os.path.basename(f)] = json.load(fh)
except (OSError, ValueError) as e:
print(f"warn: skip {f}: {e}", file=sys.stderr)
for f in sorted(glob.glob(os.path.join(tables_dir, "ro_map_*.json"))):
base = os.path.basename(f)
v8 = base[len("ro_map_"):-len(".json")].replace("_", ".")
merged_ro[v8] = base
print(f"merge-existing: {len(merged_versions)} versions, "
f"{len(merged_tables)} tables, {len(merged_ro)} ro-maps", file=sys.stderr)
donors = {}
if args.donors:
for f in glob.glob(os.path.join(args.donors, "v*.json")):
t = json.load(open(f))
donors[".".join(t["v8"].split(".")[:2])] = t
print(f"donors: {sorted(donors)}", file=sys.stderr)
index_path = args.index or os.path.join(out_root, "workspace", "node-dist-index.json")
floor = ver_key(args.all_from) if args.all_from else None
index = load_index(index_path, floor=floor) if (args.all_from or args.index) else []
index_by_v8 = {}
for e in index:
if e.get("v8"):
index_by_v8.setdefault(e["v8"], []).append(e)
if args.index_entries:
have = {(v["v8"], v["node"]) for v in merged_versions}
added = 0
for e in index:
v8 = e.get("v8")
if not v8:
continue
mk = ".".join(v8.split(".")[:2])
fname = None
for t in merged_tables.values():
if t.get("v8") == v8:
fname = next((f for f, tt in merged_tables.items() if tt is t), None)
break
if fname is None:
cand = os.path.join(tables_dir, f"v{mk.replace('.', '_')}.json")
if os.path.exists(cand):
fname = os.path.basename(cand)
if fname is None:
continue
if (v8, e["version"]) in have:
continue
p4 = [int(x) for x in v8.split(".")]
fold = next(
(t["hash"]["algorithm"] for t in merged_tables.values()
if t.get("v8") == v8 and t.get("hash")),
"left_fold" if (p4[0] > 12 or (p4[0] == 12 and p4[1] >= 2)) else "right_fold",
)
merged_versions.append({
"v8": v8,
"hash": v8_hash(fold, *p4),
"node": e["version"],
"table": fname,
})
added += 1
print(f"index-entries: +{added} releases (total {len(merged_versions)})", file=sys.stderr)
with open(os.path.join(tables_dir, "manifest.json"), "w") as f:
json.dump({"schema_version": 1,
"generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
"versions": merged_versions}, f, indent=1, sort_keys=True)
return
tags = list(args.tag) + list(args.v8_tag)
if args.all_from:
tags += tags_for_all(index, major_floor=floor)
print(f"index: {len(index)} releases >= {args.all_from}, "
f"{len(set(e['v8'] for e in index if e.get('v8')))} distinct V8 versions", file=sys.stderr)
if not tags:
ap.error("需要 --tag 或 --all-from")
tags = sorted(set(tags), key=ver_key)
versions = [] table_files = {} file_of_content = {}
tables_by_v8 = {} fold_of_v8 = {} fname_of_v8 = {} for i, tag in enumerate(tags, 1):
prefetch(tag, jobs=args.jobs)
vv = parse_v8_version(tag)
if vv is None:
print(f"skip {tag}: no v8-version.h", file=sys.stderr)
continue
print(f"[{i}/{len(tags)}] {tag} -> V8 {'.'.join(map(str, vv))}", file=sys.stderr)
maj, minor, build, patch = vv
key = f"{maj}_{minor}"
src_b = git_show(tag, "src/interpreter/bytecodes.h")
src_o = git_show(tag, "src/interpreter/bytecode-operands.h")
src_t = (
git_show(tag, "src/snapshot/serializer-deserializer.h")
or git_show(tag, "src/snapshot/serializer-common.h")
or git_show(tag, "src/snapshot/serializer.h")
)
src_serializer_h = git_show(tag, "src/snapshot/serializer.h")
src_t = (src_t or "") + "\n" + (src_serializer_h or "")
src_r = git_show(tag, "src/roots/roots.h") or git_show(tag, "src/roots.h")
src_s = git_show(tag, "src/init/heap-symbols.h") or git_show(tag, "src/heap-symbols.h")
src_d = git_show(tag, "src/objects/objects-definitions.h") or git_show(
tag, "src/objects-definitions.h"
)
src_d = "\n".join(x for x in (src_d, git_show(tag, "src/objects/object-list-macros.h")) if x)
src_rt = git_show(tag, "src/runtime/runtime.h")
src_intr = git_show(tag, "src/interpreter/interpreter-intrinsics.h")
src_code = git_show(tag, "src/objects/bytecode-array.tq") or git_show(tag, "src/objects/code.tq")
src_trusted = git_show(tag, "src/objects/trusted-object.tq")
src_fixed = git_show(tag, "src/objects/fixed-array.tq")
src_frame = git_show(tag, "src/execution/frame-constants.h") or git_show(
tag, "src/frame-constants.h"
)
src_sfi = git_show(tag, "src/objects/shared-function-info.tq")
src_scope = git_show(tag, "src/objects/scope-info.tq")
src_ba_inl = "\n".join(
x for x in (
git_show(tag, "src/objects/bytecode-array-inl.h"),
git_show(tag, "src/objects/bytecode-array.h"),
) if x
) or None
src_globals = git_show(tag, "src/common/globals.h")
if not all([src_b, src_o, src_t]):
print(f"skip {tag}: missing V8 sources", file=sys.stderr)
continue
table = {
"v8": f"{maj}.{minor}.{build}.{patch}",
"source_tag": tag,
"hash": {"algorithm": extract_hash_fold(tag)},
"header": extract_header_layout(tag),
"tagged_size": extract_tagged_size(tag),
"capture_tagged_size": extract_tagged_size(tag),
"bytecodes": extract_bytecodes(src_b),
"operand_types": extract_operand_types(src_o),
"serialization": {
"tags": extract_serialization_tags(src_t) if "enum Bytecode" in (src_t or "") else {},
"legacy": extract_legacy_tags(src_t) if maj <= 8 else {},
},
"roots": extract_roots(
src_r,
src_s,
src_d,
TORQUE_MAP_COUNT_BY_VERSION.get(key, TORQUE_MAP_COUNT),
)
if src_r
else [],
"roots_shift": 0,
"frame": apply_frame_override(tag, extract_frame_layout(src_frame)),
"scope_info": extract_scope_info(src_scope, src_globals),
"parameter_count": extract_parameter_count_semantics(src_ba_inl),
"shared_function_info": tq_layout(
{k: v for k, v in {
"shared-function-info.tq": src_sfi,
"trusted-object.tq": src_trusted,
"exposed-trusted-object.tq": src_trusted,
}.items() if v},
"SharedFunctionInfo",
extract_tagged_size(tag),
sandbox=sandbox_of(tag),
),
"runtime_names": extract_runtime_names(src_rt),
"intrinsic_names": extract_intrinsic_names(src_intr),
"bytecode_array": bytecode_array_layout(
{k: v for k, v in {
"bytecode-array.tq": src_code,
"code.tq": src_code,
"exposed-trusted-object.tq": src_trusted,
"trusted-object.tq": src_trusted,
"fixed-array.tq": src_fixed,
}.items() if v},
extract_tagged_size(tag),
sandbox=sandbox_of(tag),
),
}
if donors:
minor = f"{maj}.{minor_}" if False else f"{maj}.{minor}"
dm = donor_for(minor, donors)
if dm:
filled = _deep_fill(table, donors[dm])
if filled:
print(f" inherit {tag}: {filled} keys <- v{dm}", file=sys.stderr)
normalize_operand_types(table)
if int(table["v8"].split(".")[0]) > 8:
table.get("serialization", {}).pop("legacy", None)
content_key = hashlib.sha1(json.dumps(table, sort_keys=True).encode()).hexdigest()
if content_key not in file_of_content:
same_v8 = [
v["table"]
for v in merged_versions
if v.get("v8") == table["v8"] and v.get("table") in merged_tables
]
reused = next(
(
n
for n in same_v8
if json.dumps(merged_tables[n], sort_keys=True)
== json.dumps(table, sort_keys=True)
),
None,
)
if reused:
fname = reused
else:
fname = f"v{key}.json"
same_name_differs = (
fname in merged_tables
and json.dumps(merged_tables[fname], sort_keys=True)
!= json.dumps(table, sort_keys=True)
)
if fname in table_files or same_name_differs:
fname = f"v{key}_{tag.lstrip('v')}.json"
file_of_content[content_key] = fname
table_files[fname] = table
fname = file_of_content[content_key]
tables_by_v8[table["v8"]] = table
fname_of_v8[table["v8"]] = fname
fold_of_v8[table["v8"]] = table["hash"]["algorithm"]
entries = index_by_v8.get(table["v8"]) or [
{"version": v8_labels.get(table["v8"], tag.lstrip("v")), "v8": table["v8"]}
]
for e in entries:
e_v8 = e.get("v8") or table["v8"]
p = [int(x) for x in e_v8.split(".")]
versions.append({
"v8": e_v8,
"hash": V8_HASHES.get(tag) or v8_hash(table["hash"]["algorithm"], *p),
"node": e["version"],
"table": file_of_content[content_key],
})
if args.merge_existing:
have = {(v["v8"], v["node"]) for v in versions}
versions += [v for v in merged_versions if (v["v8"], v["node"]) not in have]
for fname, t in merged_tables.items():
if fname not in table_files:
table_files[fname] = t
if t.get("v8"):
tables_by_v8.setdefault(t["v8"], t)
fname_of_v8.setdefault(t["v8"], fname)
print(f"merge-existing: +{len(versions)} versions / {len(table_files)} tables total",
file=sys.stderr)
minor_table = {}
for fname, t in table_files.items():
minor_table.setdefault(".".join(t["v8"].split(".")[:2]), fname)
for alias_minor, target_minor in TABLE_ALIAS.items():
if target_minor in minor_table:
minor_table[alias_minor] = minor_table[target_minor]
for e in index:
e_v8 = e.get("v8")
if not e_v8:
continue
fname = None
mk = ".".join(e_v8.split(".")[:2])
if mk in TABLE_ALIAS:
fname = minor_table.get(TABLE_ALIAS[mk])
if fname is None and e_v8 in tables_by_v8:
fname = file_of_content.get(hashlib.sha1(
json.dumps(tables_by_v8[e_v8], sort_keys=True).encode()).hexdigest())
if fname is None:
fname = minor_table.get(".".join(e_v8.split(".")[:2]))
if fname is None:
print(f"warn: no table for node {e['version']} (V8 {e_v8})", file=sys.stderr)
continue
fold = fold_of_v8.get(e_v8) or (
"left_fold"
if (int(e_v8.split(".")[0]) > 12
or (int(e_v8.split(".")[0]) == 12 and int(e_v8.split(".")[1]) >= 2))
else "right_fold")
p = [int(x) for x in e_v8.split(".")]
versions.append({
"v8": e_v8,
"hash": v8_hash(fold, *p),
"node": e["version"],
"table": fname,
})
seen = set()
uniq_versions = []
for v in versions:
k = (v["v8"], v["hash"], v["node"])
if k in seen:
continue
seen.add(k)
uniq_versions.append(v)
if args.per_minor:
newest = {}
for v8 in fname_of_v8:
m = ".".join(v8.split(".")[:2])
cur = newest.get(m)
if cur is None or [int(x) for x in v8.split(".")] > [int(x) for x in cur.split(".")]:
newest[m] = v8
if args.keep_existing:
for m, v8 in newest.items():
cur = os.path.join(tables_dir, f"v{m.replace('.', '_')}.json")
if os.path.exists(cur):
fname_of_v8[v8] = os.path.basename(cur)
keep = {fname_of_v8[v8] for v8 in newest.values()}
for v in uniq_versions:
m = ".".join(v["v8"].split(".")[:2])
m = TABLE_ALIAS.get(m, m) v["table"] = fname_of_v8[newest[m]]
dropped = [f for f in table_files if f not in keep]
for f in dropped:
del table_files[f]
print(f"per-minor: {len(keep)} tables kept, {len(dropped)} dropped", file=sys.stderr)
manifest = {
"schema_version": 1,
"generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
"versions": uniq_versions,
}
with open(os.path.join(tables_dir, "manifest.json"), "w") as f:
json.dump(manifest, f, indent=1, sort_keys=True)
for fname, table in table_files.items():
dest = os.path.join(tables_dir, fname)
if args.keep_existing and os.path.exists(dest):
continue with open(dest, "w") as f:
json.dump(table, f, indent=1, sort_keys=True)
ro_files = {}
if args.ro_maps:
per_minor_pick = {}
for e in index:
v8 = e.get("v8") or ""
if not v8 or v8 in per_minor_pick:
continue
cand = os.path.join(args.ro_maps, f"ro-map-{e['version'].lstrip('v')}.json")
if not os.path.exists(cand):
continue
per_minor_pick[v8] = cand
seen_ro = {}
for v8str, path in sorted(per_minor_pick.items()):
text = open(path, encoding="utf-8").read()
h = hashlib.sha1(text.encode()).hexdigest()
if h in seen_ro:
ro_files[v8str] = seen_ro[h]
continue
fname = f"ro_map_{v8str.replace('.', '_')}.json"
seen_ro[h] = fname
ro_files[v8str] = fname
with open(os.path.join(tables_dir, fname), "w", encoding="utf-8") as f:
f.write(text)
print(f"ro-maps: {len(ro_files)} V8 versions", file=sys.stderr)
for v8str, fname in merged_ro.items():
ro_files.setdefault(v8str, fname)
lines = ["// 本文件由 scripts/codegen.py 生成，请勿手改。"]
lines.append(f"pub static MANIFEST_JSON: &str = include_str!(\"../tables/manifest.json\");")
lines.append("pub static EMBEDDED_TABLE_FILES: &[(&str, &str)] = &[")
for fname in sorted(table_files):
lines.append(f" (\"{fname}\", include_str!(\"../tables/{fname}\")),")
lines.append("];")
with open(os.path.join(out_root, "src", "tables_embed.rs"), "w") as f:
f.write("\n".join(lines) + "\n")
if ro_files:
rlines = ["// 本文件由 scripts/codegen.py 生成，请勿手改。",
"pub static RO_MAPS: &[(&str, &str)] = &["]
for mk in sorted(ro_files, key=lambda s: tuple(int(x) for x in s.split("."))):
rlines.append(
f' ("{mk}", include_str!("../tables/{ro_files[mk]}")),')
rlines.append("];")
with open(os.path.join(out_root, "src", "ro_embed.rs"), "w") as f:
f.write("\n".join(rlines) + "\n")
print(f"tables: {len(table_files)} files, {len(uniq_versions)} hash entries")
if __name__ == "__main__":
main()