cf8c4cccc1
The front end hardcoded /opt/homebrew/opt/llvm and lib/libclang.dylib, so
on Linux it died before parsing anything:
FileNotFoundError: [Errno 2] No such file or directory:
'/opt/homebrew/opt/llvm/bin/clang'
PIPE_LLVM could move the prefix but not the file name, so no value of it
made the tool run there.
Discovery is now per platform: Homebrew's LLVM on macOS, the system prefix
elsewhere, PIPE_LLVM still winning over both. The library is looked up as
.dylib or .so, exact name first and then highest version, and under
lib/*-linux-gnu as well, because distributions ship
libclang.so.<major>.<minor> without always adding the unversioned symlink
and Debian keeps it in a multiarch subdirectory. -print-resource-dir falls
back to clang on PATH when the prefix ships only the library. When nothing
matches, the conventional name is handed to clang.cindex so its own error
names the library rather than a path this code invented.
_sdk_root already degrades to None off macOS, since xcrun is absent.
test/cpp_front_test.dart looked for the same Homebrew dylib, so on Linux
it reported "All tests skipped" rather than failing. That is why the gap
went unnoticed: the suite was green because it ran nothing. Its detection
now mirrors the front end's, and on this Linux machine the four cases go
from skipped to run and passing.
Verified end to end against an unrelated 87-translation-unit C++ project:
87/87 parsed, 0 TUs failed, 0 clang errors, 26 state enums and 879
emissions recovered.
Both suites fail test/stdout_json_test.dart under the parallel runner on
this machine, pristine main included, and it passes in isolation on both
branches; a serial run failed a different test instead. Pre-existing and
unrelated.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
679 lines
29 KiB
Python
679 lines
29 KiB
Python
#!/usr/bin/env python3
|
|
"""Experimental C++ front end: writes a pipe state model from clang's AST.
|
|
|
|
Usage:
|
|
tool/cpp_front.py <project-root> [--build-dir DIR] [--include src,include]
|
|
[--limit N] [--jobs N] [-o model.json]
|
|
pipe check --model model.json
|
|
|
|
pipe's rules read only the state model, so this front end never touches them.
|
|
It uses libclang's *resolved* AST: every enumerator reference is linked to its
|
|
EnumConstantDecl, so identity is exact rather than matched by name.
|
|
|
|
What it maps (scoped enums only, `enum class`, declared in the project):
|
|
|
|
union an `enum class`; variants are its enumerators
|
|
emission a reference that PRODUCES the value: assigned to a field or
|
|
variable, stored into an atomic, passed, returned, initialised
|
|
observation a reference that EXAMINES the value: a `case` label, `==`, `!=`
|
|
(these fill pipe's "rendered" set: something branches on it)
|
|
swallowed a member writes a busy enumerator (Starting, Loading, ...) to a
|
|
failure field, then has a try whose catch writes no state and does not
|
|
rethrow (loading-not-reset)
|
|
|
|
Consequences worth knowing before reading the findings:
|
|
- `emitted-not-rendered` here means "produced, but nothing ever branches on or
|
|
compares it". A value that reaches the user through a lookup table or
|
|
reflection (magic_enum) is invisible to this, so treat those as leads.
|
|
- `rendered-not-emitted` means "compared against, but never produced anywhere":
|
|
a dead branch, unless the value comes from a cast (`static_cast<E>(n)`), which
|
|
is recorded as an indirect emission so the rule suppresses itself.
|
|
- The rule messages still speak of "UI"; for C++ read "nothing observes it".
|
|
|
|
Requires libclang and its Python bindings (found automatically): Homebrew's
|
|
LLVM on macOS, the system LLVM elsewhere. Set PIPE_LLVM to point at another
|
|
prefix.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
from collections import Counter, defaultdict
|
|
from concurrent.futures import ProcessPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
|
|
def _llvm_prefix() -> Path:
|
|
"""Where libclang lives. PIPE_LLVM wins; otherwise Homebrew's LLVM on macOS
|
|
and the system prefix elsewhere, because Linux has no /opt/homebrew."""
|
|
env = os.environ.get("PIPE_LLVM")
|
|
if env:
|
|
return Path(env)
|
|
return Path("/opt/homebrew/opt/llvm" if sys.platform == "darwin" else "/usr")
|
|
|
|
|
|
LLVM = _llvm_prefix()
|
|
|
|
|
|
def _libclang_path() -> str:
|
|
"""The shared library, by platform and possibly versioned.
|
|
|
|
Linux distributions name it libclang.so.<major>.<minor> and do not always
|
|
ship the unversioned symlink, and Debian/Ubuntu put it under a multiarch
|
|
subdirectory -- so a single fixed name finds nothing on most of them.
|
|
"""
|
|
suffix = "dylib" if sys.platform == "darwin" else "so"
|
|
roots = [LLVM / "lib", *(Path(p) for p in glob.glob(str(LLVM / "lib/*-linux-gnu")))]
|
|
for root in roots:
|
|
exact = root / f"libclang.{suffix}"
|
|
if exact.exists():
|
|
return str(exact)
|
|
for root in roots:
|
|
# Highest version first: libclang.so.22.1 before libclang.so.21.1.
|
|
versioned = sorted(glob.glob(str(root / f"libclang.{suffix}*")), reverse=True)
|
|
if versioned:
|
|
return versioned[0]
|
|
# Nothing found: hand clang.cindex the conventional name so its own error
|
|
# names the library rather than a path this function invented.
|
|
return str(LLVM / f"lib/libclang.{suffix}")
|
|
|
|
|
|
def _load_cindex():
|
|
for candidate in sorted(glob.glob(str(LLVM / "lib/python3.*/site-packages")), reverse=True):
|
|
if candidate not in sys.path:
|
|
sys.path.insert(0, candidate)
|
|
import clang.cindex as cindex # noqa: E402
|
|
|
|
# Pool workers are reused, and libclang may be configured only once.
|
|
if not cindex.Config.loaded:
|
|
cindex.Config.set_library_file(_libclang_path())
|
|
return cindex
|
|
|
|
|
|
_SDK = None
|
|
|
|
BUSY = re.compile(
|
|
r"^(Loading|Starting|StartingUp|Initializing|Initialising|Connecting|Stopping|"
|
|
r"ShuttingDown|Busy|Pending|InProgress|Processing|Saving|Updating|Creating|"
|
|
r"Syncing|Indexing|Warming|WarmingUp|Draining|Migrating|Recovering)$"
|
|
)
|
|
|
|
# Kinds that only wrap an expression; the classifier looks through them.
|
|
_TRANSPARENT = {
|
|
"UNEXPOSED_EXPR",
|
|
"PAREN_EXPR",
|
|
"CSTYLE_CAST_EXPR",
|
|
"CXX_FUNCTIONAL_CAST_EXPR",
|
|
"CONDITIONAL_OPERATOR",
|
|
"INIT_LIST_EXPR",
|
|
}
|
|
|
|
|
|
def _clang_binary() -> str:
|
|
"""The driver used only for -print-resource-dir. Under a system prefix
|
|
LLVM/bin/clang exists, but a prefix that ships just the library does not
|
|
have one, so fall back to whatever is on PATH."""
|
|
candidate = LLVM / "bin/clang"
|
|
if candidate.exists():
|
|
return str(candidate)
|
|
return shutil.which("clang") or shutil.which("clang++") or str(candidate)
|
|
|
|
|
|
def _resource_dir() -> str:
|
|
return subprocess.check_output([_clang_binary(), "-print-resource-dir"], text=True).strip()
|
|
|
|
|
|
def _sdk_root() -> str | None:
|
|
"""Apple's compiler finds the macOS SDK itself; libclang needs it spelled out,
|
|
or every standard header (<chrono>, sqlite3.h) is missing."""
|
|
try:
|
|
return subprocess.check_output(["xcrun", "--show-sdk-path"], text=True).strip()
|
|
except (OSError, subprocess.CalledProcessError):
|
|
return None
|
|
|
|
|
|
def _args_for(entry: dict, resource_dir: str) -> list[str]:
|
|
"""Compiler flags only: drop wrappers (ccache), the compiler, the source,
|
|
and output/dependency options that make no sense for a syntax-only parse."""
|
|
args = entry.get("arguments") or shlex.split(entry["command"])
|
|
while args and (os.path.basename(args[0]) in ("ccache", "sccache", "distcc")
|
|
or not args[0].startswith("-") and args[0] != entry["file"]
|
|
and os.path.basename(args[0]) in ("c++", "g++", "clang++", "cc", "gcc", "clang")):
|
|
args = args[1:]
|
|
out: list[str] = []
|
|
skip = False
|
|
directory = entry["directory"]
|
|
# Include paths are relative to the build directory, but libclang resolves
|
|
# them against this process's working directory: made absolute, or headers
|
|
# like third_party/... silently go missing.
|
|
path_flags = ("-I", "-isystem", "-iquote", "-idirafter", "-include")
|
|
absolute = []
|
|
pending = None
|
|
for arg in args:
|
|
if pending is not None:
|
|
absolute.append(arg if os.path.isabs(arg) else os.path.normpath(os.path.join(directory, arg)))
|
|
pending = None
|
|
continue
|
|
if arg in path_flags:
|
|
absolute.append(arg)
|
|
pending = arg
|
|
continue
|
|
for flag in path_flags:
|
|
if arg.startswith(flag) and len(arg) > len(flag) and flag != "-include":
|
|
value = arg[len(flag):]
|
|
if not os.path.isabs(value):
|
|
arg = flag + os.path.normpath(os.path.join(directory, value))
|
|
break
|
|
absolute.append(arg)
|
|
args = absolute
|
|
for arg in args:
|
|
if skip:
|
|
skip = False
|
|
continue
|
|
if arg in ("-o", "-MF", "-MT", "-MQ"):
|
|
skip = True
|
|
continue
|
|
if arg in ("-c", "-MD", "-MMD", "-fdiagnostics-color=always") or \
|
|
arg.startswith(("-o", "-MF", "-MT", "-MQ")):
|
|
continue
|
|
if arg == entry["file"] or not arg.startswith("-") and arg.endswith((".cpp", ".cc", ".cxx")):
|
|
continue
|
|
out.append(arg)
|
|
out += ["-resource-dir", resource_dir, "-Wno-everything", "-fsyntax-only"]
|
|
sdk = _SDK
|
|
if sdk and "-isysroot" not in out:
|
|
out += ["-isysroot", sdk]
|
|
return out
|
|
|
|
|
|
def _qualified(cursor) -> str:
|
|
parts = []
|
|
c = cursor
|
|
while c is not None and c.kind.name != "TRANSLATION_UNIT":
|
|
if c.spelling:
|
|
parts.append(c.spelling)
|
|
c = c.semantic_parent
|
|
return "::".join(reversed(parts))
|
|
|
|
|
|
def _enclosing(stack, kinds):
|
|
for c in reversed(stack):
|
|
if c.kind.name in kinds:
|
|
return c
|
|
return None
|
|
|
|
|
|
def _operator(cursor) -> str | None:
|
|
try:
|
|
op = cursor.binary_operator
|
|
return op.name if op is not None else None
|
|
except AttributeError:
|
|
return None
|
|
|
|
|
|
def _excerpt(cursor, limit: int = 120) -> str:
|
|
"""A bounded, one-line excerpt of [cursor]'s source that is safe to quote
|
|
to an agent: analysed source is untrusted, so string and character literals
|
|
are elided (as pipe's Dart front end does) and the result is capped."""
|
|
parts = []
|
|
for token in cursor.get_tokens():
|
|
text = token.spelling
|
|
if token.kind.name == "LITERAL" and text[:1] in ("\"", "'") or \
|
|
token.kind.name == "LITERAL" and re.match(r'^[uUL]?8?R?["\']', text):
|
|
text = "'…'"
|
|
if token.kind.name == "COMMENT":
|
|
continue
|
|
parts.append(text)
|
|
collapsed = re.sub(r"\s+", " ", " ".join(parts)).strip()
|
|
if not collapsed:
|
|
return "<empty catch>"
|
|
return collapsed if len(collapsed) <= limit else collapsed[: limit - 3] + "..."
|
|
|
|
|
|
def _top_level_try(stack) -> bool:
|
|
"""True when the try sits directly in its function's body, not nested in a
|
|
loop, branch, block or another try."""
|
|
for i in range(len(stack) - 1, -1, -1):
|
|
kind = stack[i].kind.name
|
|
if kind in ("CXX_METHOD", "FUNCTION_DECL", "CONSTRUCTOR", "LAMBDA_EXPR"):
|
|
# stack[i+1] is the body; the try must be its direct child.
|
|
return len(stack) - 1 == i + 1 and stack[i + 1].kind.name == "COMPOUND_STMT"
|
|
if kind == "CXX_TRY_STMT":
|
|
return False
|
|
return False
|
|
|
|
|
|
def _field_owner(expr, cindex):
|
|
"""For `x.state_` / `state_` / `this->state_`: the class declaring the field."""
|
|
for node in expr.walk_preorder():
|
|
if node.kind.name == "MEMBER_REF_EXPR":
|
|
ref = node.referenced
|
|
if ref is not None and ref.kind.name == "FIELD_DECL":
|
|
return ref.semantic_parent, ref.spelling
|
|
return None, None
|
|
|
|
|
|
def analyse_tu(task):
|
|
"""Parse one translation unit; return the facts located in project files."""
|
|
file, directory, args, root, includes = task
|
|
cindex = _load_cindex()
|
|
index = cindex.Index.create()
|
|
try:
|
|
tu = index.parse(
|
|
os.path.normpath(os.path.join(directory, file)),
|
|
args=args,
|
|
options=cindex.TranslationUnit.PARSE_INCOMPLETE,
|
|
)
|
|
except cindex.TranslationUnitLoadError as error:
|
|
return {"file": file, "error": str(error)}
|
|
|
|
root = os.path.realpath(root)
|
|
prefixes = tuple(os.path.join(root, p) + os.sep for p in includes)
|
|
|
|
def in_project(cursor) -> str | None:
|
|
loc = cursor.location
|
|
if loc.file is None:
|
|
return None
|
|
path = os.path.realpath(loc.file.name)
|
|
if not path.startswith(prefixes):
|
|
return None
|
|
return os.path.relpath(path, root)
|
|
|
|
enums = {}
|
|
refs = []
|
|
busy_tries = []
|
|
|
|
def classify(stack, ref):
|
|
"""Return (role, owner, field, kind) for an enumerator reference."""
|
|
child = ref
|
|
for parent in reversed(stack):
|
|
kind = parent.kind.name
|
|
if kind in _TRANSPARENT:
|
|
child = parent
|
|
continue
|
|
if kind == "CASE_STMT":
|
|
first = next(iter(parent.get_children()), None)
|
|
if first is not None and first.hash == child.hash:
|
|
return ("observe", None, None, "case")
|
|
return ("produce", None, None, "other")
|
|
if kind == "BINARY_OPERATOR":
|
|
op = _operator(parent)
|
|
children = list(parent.get_children())
|
|
if op in ("EQ", "NE"):
|
|
return ("observe", None, None, "compare")
|
|
if op == "Assign" and len(children) == 2 and children[1].hash == child.hash:
|
|
owner, field = _field_owner(children[0], cindex)
|
|
return ("produce", owner, field, "assign")
|
|
return ("produce", None, None, "other")
|
|
if kind == "CALL_EXPR" and parent.spelling in ("store", "exchange"):
|
|
owner, field = _field_owner(parent, cindex)
|
|
return ("produce", owner, field, "assign")
|
|
if kind == "MEMBER_REF":
|
|
ref_decl = parent.referenced
|
|
if ref_decl is not None and ref_decl.kind.name == "FIELD_DECL":
|
|
return ("produce", ref_decl.semantic_parent, ref_decl.spelling, "initial")
|
|
if kind == "FIELD_DECL":
|
|
return ("produce", parent.semantic_parent, parent.spelling, "initial")
|
|
if kind in ("CALL_EXPR", "RETURN_STMT", "VAR_DECL", "COMPOUND_STMT",
|
|
"IF_STMT", "CXX_METHOD", "FUNCTION_DECL", "CONSTRUCTOR"):
|
|
return ("produce", None, None, "other")
|
|
child = parent
|
|
return ("produce", None, None, "other")
|
|
|
|
def member_name(stack):
|
|
fn = _enclosing(stack, {"CXX_METHOD", "FUNCTION_DECL", "CONSTRUCTOR",
|
|
"DESTRUCTOR", "FUNCTION_TEMPLATE", "LAMBDA_EXPR"})
|
|
if fn is None:
|
|
return "<top>", None
|
|
if fn.kind.name == "LAMBDA_EXPR":
|
|
outer = _enclosing(stack[: stack.index(fn)], {"CXX_METHOD", "FUNCTION_DECL", "CONSTRUCTOR"})
|
|
name = outer.spelling if outer is not None else "<lambda>"
|
|
return f"{name}::<lambda>", outer
|
|
return fn.spelling, fn
|
|
|
|
def scan_try(stack, try_cursor, rel):
|
|
"""A try inside a member that wrote a busy value earlier."""
|
|
body = list(try_cursor.get_children())
|
|
handlers = [c for c in body if c.kind.name == "CXX_CATCH_STMT"]
|
|
swallowed = []
|
|
for handler in handlers:
|
|
writes_state = False
|
|
rethrows = False
|
|
for node in handler.walk_preorder():
|
|
if node.kind.name == "CXX_THROW_EXPR":
|
|
rethrows = True
|
|
if node.kind.name == "DECL_REF_EXPR":
|
|
ref = node.referenced
|
|
if ref is not None and ref.kind.name == "ENUM_CONSTANT_DECL":
|
|
writes_state = True
|
|
if node.kind.name in ("CALL_EXPR",) and re.search(
|
|
r"(set|mark|fail|error|transition|reset|clear)", node.spelling or "", re.I
|
|
):
|
|
writes_state = True
|
|
if not writes_state and not rethrows:
|
|
swallowed.append((handler.location.line, _excerpt(handler), handler))
|
|
return swallowed
|
|
|
|
def visit(cursor, stack, busy_seen):
|
|
rel = in_project(cursor)
|
|
if rel is None and cursor.kind.name != "TRANSLATION_UNIT":
|
|
return
|
|
kind = cursor.kind.name
|
|
if kind == "ENUM_DECL" and cursor.is_scoped_enum() and cursor.is_definition():
|
|
name = _qualified(cursor)
|
|
enums[name] = {
|
|
"name": name,
|
|
"declaredIn": rel,
|
|
"line": cursor.location.line,
|
|
"variants": [c.spelling for c in cursor.get_children()
|
|
if c.kind.name == "ENUM_CONSTANT_DECL"],
|
|
}
|
|
elif kind == "DECL_REF_EXPR":
|
|
ref = cursor.referenced
|
|
if ref is not None and ref.kind.name == "ENUM_CONSTANT_DECL":
|
|
parent_enum = ref.semantic_parent
|
|
if parent_enum is not None and parent_enum.is_scoped_enum():
|
|
role, owner, field, how = classify(stack, cursor)
|
|
member, fn = member_name(stack)
|
|
klass = _enclosing(stack, {"CLASS_DECL", "STRUCT_DECL", "CLASS_TEMPLATE"})
|
|
if fn is not None and fn.semantic_parent is not None and \
|
|
fn.semantic_parent.kind.name in ("CLASS_DECL", "STRUCT_DECL", "CLASS_TEMPLATE"):
|
|
klass = fn.semantic_parent
|
|
record = {
|
|
"enum": _qualified(parent_enum),
|
|
"variant": ref.spelling,
|
|
"role": role,
|
|
"how": how,
|
|
"owner": _qualified(owner) if owner is not None else None,
|
|
"field": field,
|
|
"member": member,
|
|
"memberOwner": _qualified(klass) if klass is not None else None,
|
|
"path": rel,
|
|
"line": cursor.location.line,
|
|
"col": cursor.location.column,
|
|
}
|
|
refs.append(record)
|
|
# Only the method's OWN state counts: `ev.stage = Connecting` on a
|
|
# local event object handed to a callback is not state this
|
|
# class can leave stuck.
|
|
if role == "produce" and how == "assign" and BUSY.match(ref.spelling) \
|
|
and owner is not None and klass is not None \
|
|
and _qualified(owner) == _qualified(klass):
|
|
busy_seen.append(record)
|
|
elif kind == "CXX_STATIC_CAST_EXPR" or kind == "CSTYLE_CAST_EXPR":
|
|
target = cursor.type.get_canonical()
|
|
decl = target.get_declaration()
|
|
if decl is not None and decl.kind.name == "ENUM_DECL" and decl.is_scoped_enum():
|
|
children = list(cursor.get_children())
|
|
if children and children[-1].type.get_canonical().kind.name != "ENUM":
|
|
refs.append({
|
|
"enum": _qualified(decl), "variant": None, "role": "cast", "how": "cast",
|
|
"owner": None, "field": None, "member": member_name(stack)[0],
|
|
"memberOwner": None, "path": rel, "line": cursor.location.line,
|
|
"col": cursor.location.column,
|
|
})
|
|
elif kind == "CXX_TRY_STMT" and busy_seen and _top_level_try(stack):
|
|
# (a top-level try of the method: nested tries are local fallbacks,
|
|
# like a JSON parse guarded with `catch (...) { skip = false; }`)
|
|
member, fn = member_name(stack)
|
|
for busy in busy_seen:
|
|
if busy["member"] != member:
|
|
continue
|
|
# `member#index` among every catch of the enclosing function, as
|
|
# the Dart front end does: stable when code outside it moves.
|
|
catches = [c for c in fn.walk_preorder() if c.kind.name == "CXX_CATCH_STMT"] \
|
|
if fn is not None else []
|
|
for line, handler, clause in scan_try(stack, cursor, rel):
|
|
index = next((i for i, c in enumerate(catches) if c == clause), -1)
|
|
busy_tries.append({
|
|
"owner": busy["owner"], "member": member, "path": rel,
|
|
"line": line, "handler": handler, "site": f"{member}#{index}",
|
|
})
|
|
break
|
|
new_busy = busy_seen
|
|
if kind in ("CXX_METHOD", "FUNCTION_DECL", "CONSTRUCTOR", "LAMBDA_EXPR"):
|
|
new_busy = []
|
|
stack.append(cursor)
|
|
for child in cursor.get_children():
|
|
visit(child, stack, new_busy)
|
|
stack.pop()
|
|
|
|
visit(tu.cursor, [], [])
|
|
errors = [f"{d.location.file}:{d.location.line}: {d.spelling}"
|
|
for d in tu.diagnostics if d.severity >= 3]
|
|
return {"file": file, "enums": list(enums.values()), "refs": refs, "tries": busy_tries,
|
|
"errors": len(errors), "errorSamples": errors[:3]}
|
|
|
|
|
|
def build_model(root: Path, results: list[dict], includes: list[str]) -> dict:
|
|
enums: dict[str, dict] = {}
|
|
seen_refs = set()
|
|
refs = []
|
|
tries = {}
|
|
failed = []
|
|
for result in results:
|
|
if "error" in result:
|
|
failed.append(result["file"])
|
|
continue
|
|
for enum in result["enums"]:
|
|
enums.setdefault(enum["name"], enum)
|
|
for ref in result["refs"]:
|
|
key = (ref["path"], ref["line"], ref["col"], ref["enum"], ref["variant"])
|
|
if key in seen_refs:
|
|
continue
|
|
seen_refs.add(key)
|
|
refs.append(ref)
|
|
for t in result["tries"]:
|
|
tries.setdefault((t["path"], t["line"]), t)
|
|
|
|
by_enum = defaultdict(list)
|
|
for ref in refs:
|
|
by_enum[ref["enum"]].append(ref)
|
|
|
|
unions, coverage, emissions, renders = [], [], [], []
|
|
for name, enum in sorted(enums.items()):
|
|
if len(enum["variants"]) < 2:
|
|
continue
|
|
erefs = by_enum.get(name, [])
|
|
owners = Counter(r["owner"] for r in erefs if r["role"] == "produce" and r["owner"])
|
|
# The class whose field holds this state; without one it is a plain value
|
|
# type (a mode flag, a return code), not state anyone could leave stale.
|
|
if not owners:
|
|
continue
|
|
owner = owners.most_common(1)[0][0]
|
|
variants = enum["variants"]
|
|
emitted, initial, observed = set(), set(), set()
|
|
observe_sites = 0
|
|
emit_sites = 0
|
|
for r in erefs:
|
|
if r["role"] == "observe":
|
|
observed.add(r["variant"])
|
|
observe_sites += 1
|
|
renders.append({
|
|
"path": r["path"], "line": r["line"], "widget": r["how"],
|
|
"stateType": name,
|
|
"branches": [{"variant": r["variant"], "line": r["line"], "trivial": False}],
|
|
})
|
|
elif r["role"] in ("produce", "cast"):
|
|
emit_sites += 1
|
|
if r["variant"]:
|
|
emitted.add(r["variant"])
|
|
if r["how"] == "initial":
|
|
initial.add(r["variant"])
|
|
emissions.append({
|
|
"path": r["path"], "line": r["line"], "owner": owner,
|
|
"member": r["member"], "kind": "assign" if r["how"] == "assign" else "emit",
|
|
"confidence": "exact" if r["role"] == "produce" else "indirect",
|
|
"variants": [r["variant"]] if r["variant"] else [],
|
|
"snippet": f"{r['field'] or ''} = {name.split('::')[-1]}::{r['variant']}"
|
|
if r["variant"] else f"static_cast<{name.split('::')[-1]}>(...)",
|
|
# A class holds several enums: name the union so a cast to one
|
|
# does not make "never emitted" unprovable for the others.
|
|
"stateType": name,
|
|
})
|
|
union = {
|
|
"name": name, "declaredIn": enum["declaredIn"], "style": "enumUnion",
|
|
"owner": owner, "ownerKind": "Cubit",
|
|
"variants": [{"name": v, "declaredIn": enum["declaredIn"]} for v in variants],
|
|
}
|
|
unions.append(union)
|
|
coverage.append({
|
|
"union": name, "owner": owner, "style": "enumUnion",
|
|
"declared": sorted(variants), "emitted": sorted(emitted), "initial": sorted(initial),
|
|
"rendered": sorted(observed), "renderedInBuilder": [], "trivialInBuilder": [],
|
|
"emissionSites": emit_sites, "renderSites": observe_sites,
|
|
})
|
|
|
|
owners = {u["owner"] for u in unions}
|
|
swallowed = []
|
|
for (path, line), t in sorted(tries.items()):
|
|
if t["owner"] not in owners:
|
|
continue
|
|
swallowed.append({
|
|
"owner": t["owner"], "member": t["member"], "path": path, "line": line,
|
|
"site": t["site"], "handler": t["handler"],
|
|
})
|
|
|
|
# Files outside an owner's own declaration that mention it: pipe uses these
|
|
# to tell a dormant (unwired) union from a partly observed one.
|
|
references = defaultdict(set)
|
|
for r in refs:
|
|
if r["role"] == "observe" and r["enum"] in {u["name"] for u in unions}:
|
|
owner = next(u["owner"] for u in unions if u["name"] == r["enum"])
|
|
references[owner].add(r["path"])
|
|
|
|
return {
|
|
"root": str(root),
|
|
"directories": includes,
|
|
"frontEnd": {"name": "pipe-cpp-front (experimental)", "failedTranslationUnits": failed},
|
|
"unions": unions,
|
|
"eventUnions": [],
|
|
"emissions": emissions,
|
|
"renders": renders,
|
|
"handlers": [],
|
|
"dispatches": [],
|
|
"coverage": coverage,
|
|
"ownerReferences": {k: sorted(v) for k, v in references.items()},
|
|
"flagStates": [],
|
|
"riverpod": {"providers": [], "sites": []},
|
|
"constructedByFile": {},
|
|
"swallowedFailures": swallowed,
|
|
"unsettledFlags": [],
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
parser.add_argument("root")
|
|
parser.add_argument("--build-dir", help="directory holding compile_commands.json")
|
|
parser.add_argument("--include", default="src,include",
|
|
help="project subdirectories whose code is analysed")
|
|
parser.add_argument("--limit", type=int)
|
|
parser.add_argument("--match", help="only translation units whose path contains this")
|
|
parser.add_argument("--jobs", type=int, default=os.cpu_count())
|
|
parser.add_argument("-o", "--output", default="-")
|
|
args = parser.parse_args()
|
|
|
|
root = Path(args.root).resolve()
|
|
db_dir = Path(args.build_dir) if args.build_dir else root
|
|
db_path = db_dir / "compile_commands.json"
|
|
if not db_path.is_file():
|
|
print(f"cpp-front: no compile database at {db_path}; generate one "
|
|
f"(meson/CMake write it to the build dir) and pass --build-dir",
|
|
file=sys.stderr)
|
|
return 2
|
|
database = json.loads(db_path.read_text())
|
|
includes = [p for p in args.include.split(",") if p]
|
|
resource_dir = _resource_dir()
|
|
global _SDK
|
|
_SDK = _sdk_root()
|
|
|
|
tasks = []
|
|
seen = set()
|
|
for entry in database:
|
|
# Resolved on both sides: a symlinked path (macOS /var -> /private/var)
|
|
# otherwise lands "outside" the resolved root and the TU is skipped.
|
|
path = os.path.realpath(os.path.join(entry["directory"], entry["file"]))
|
|
if path in seen or not path.endswith((".cpp", ".cc", ".cxx")):
|
|
continue
|
|
rel = os.path.relpath(path, root)
|
|
if not any(rel.startswith(p + os.sep) for p in includes):
|
|
continue
|
|
if args.match and args.match not in rel:
|
|
continue
|
|
seen.add(path)
|
|
tasks.append((entry["file"], entry["directory"], _args_for(entry, resource_dir),
|
|
str(root), includes))
|
|
if args.limit:
|
|
tasks = tasks[: args.limit]
|
|
if not tasks:
|
|
# An empty model reads as a clean project; refuse instead.
|
|
print(f"cpp-front: no translation unit under {', '.join(includes)} in "
|
|
f"{db_dir / 'compile_commands.json'}; nothing was analysed", file=sys.stderr)
|
|
return 2
|
|
|
|
results = []
|
|
with ProcessPoolExecutor(max_workers=args.jobs) as pool:
|
|
futures = [pool.submit(analyse_tu, t) for t in tasks]
|
|
for i, future in enumerate(as_completed(futures), 1):
|
|
results.append(future.result())
|
|
if i % 25 == 0 or i == len(tasks):
|
|
print(f"cpp-front: {i}/{len(tasks)} translation units", file=sys.stderr)
|
|
|
|
model = build_model(root, results, includes)
|
|
|
|
# Sources the build does not compile (optional backends, benchmarks) were
|
|
# never analysed, so evidence in them is missing. Recorded as unresolved with
|
|
# the names they mention, pipe lowers confidence for any finding naming them.
|
|
unanalysed = []
|
|
names: set[str] = set()
|
|
for include in includes:
|
|
for path in sorted((root / include).rglob("*")):
|
|
if path.suffix not in (".cpp", ".cc", ".cxx") or os.path.realpath(path) in seen:
|
|
continue
|
|
unanalysed.append(os.path.relpath(path, root))
|
|
names.update(re.findall(r"[A-Za-z_][A-Za-z0-9_]*", path.read_text(errors="replace")))
|
|
failed_files = [os.path.normpath(f) for f in model["frontEnd"]["failedTranslationUnits"]]
|
|
# The same coverage block the Dart front end writes: every report built from
|
|
# this model states how much of the project it covers.
|
|
model["scanCoverage"] = {
|
|
"discovered": len(tasks) + len(unanalysed),
|
|
"selected": len(tasks),
|
|
"resolved": len(tasks) - len(failed_files),
|
|
"failed": len(failed_files),
|
|
"skipped": len(unanalysed),
|
|
}
|
|
if unanalysed or failed_files:
|
|
files = unanalysed + failed_files
|
|
model["unresolved"] = {
|
|
"files": files,
|
|
"analysed": len(tasks) - len(failed_files),
|
|
"first": f"{files[0]}: not compiled by this build, so not analysed",
|
|
"names": sorted(names),
|
|
}
|
|
failed = model["frontEnd"]["failedTranslationUnits"]
|
|
errors = sum(r.get("errors", 0) for r in results)
|
|
for r in results:
|
|
for sample in r.get("errorSamples", [])[:1]:
|
|
print(f"cpp-front: clang error in {r['file']}: {sample}", file=sys.stderr)
|
|
print(f"cpp-front: {len(model['unions'])} state enums, {len(model['emissions'])} "
|
|
f"emissions, {len(model['renders'])} observations, "
|
|
f"{len(model['swallowedFailures'])} swallowed failures; "
|
|
f"{len(failed)} TU(s) failed, {errors} clang error(s)", file=sys.stderr)
|
|
text = json.dumps(model, indent=2)
|
|
if args.output == "-":
|
|
print(text)
|
|
else:
|
|
Path(args.output).write_text(text)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|