「并行与增量编译」

# 全量 vs 增量: 模拟重编译工作量
def full_build(units):
    return sum(u["cost"] for u in units)

def incremental_build(units, changed):
    return sum(u["cost"] for u in units if u["name"] in changed)

units = [{"name": f"m{i}", "cost": 3} for i in range(100)]
print("全量编译:", full_build(units), "单位工作量")
print("改动 2 个文件的增量编译:", incremental_build(units, {"m3", "m7"}), "单位工作量")

def topo_sort(adj):
    """Kahn 算法: 每次取出一个入度为 0 的节点."""
    from collections import deque
    indeg = {n: 0 for n in adj}
    for edges in adj.values():
        for e in edges:
            indeg[e] = indeg.get(e, 0) + 1
    q = deque([n for n, d in indeg.items() if d == 0])
    order = []
    while q:
        n = q.popleft()
        order.append(n)
        for e in adj.get(n, ()):
            indeg[e] -= 1
            if indeg[e] == 0:
                q.append(e)
    return order

# 边 core->ui / core->main 表示 ui、main 都依赖 core
graph = {"core": {"ui", "main"}, "ui": {"main"}, "main": set()}
print("编译顺序:", topo_sort(graph))

# 依赖图可视化
def render_dag(adj):
    lines = ["digraph deps {"]
    for u, vs in adj.items():
        for v in vs:
            lines.append(f'  "{u}" -> "{v}"')
    lines.append("}")
    return "\n".join(lines)

print(render_dag(graph))

import threading
from collections import deque

def parallel_build(adj, workers=4):
    """adj[u] = 依赖 u 的模块集合; 边指向依赖者."""
    indeg = {n: 0 for n in adj}         # indeg[v] = v 的未完成前置依赖数
    for u, vs in adj.items():
        for v in vs:
            indeg[v] = indeg.get(v, 0) + 1
    ready = deque(n for n, d in indeg.items() if d == 0)
    done = []
    lock = threading.Lock()

    def worker():
        while True:
            with lock:
                if not ready:
                    return
                n = ready.popleft()
            import time; time.sleep(0.05)       # 模拟编译耗时
            with lock:
                done.append(n)
                for u in adj.get(n, ()):        # n 完成后解锁依赖它的模块
                    indeg[u] -= 1
                    if indeg[u] == 0:
                        ready.append(u)

    threads = [threading.Thread(target=worker) for _ in range(workers)]
    for t in threads: t.start()
    for t in threads: t.join()
    return done

# 与上节同一依赖图: core 先编译, 再 ui, 最后 main
print("并行编译完成顺序:", parallel_build(
    {"core": {"ui", "main"}, "ui": {"main"}, "main": set()}
))

import hashlib, json

def fingerprint(source, dep_interfaces, flags, compiler_version):
    payload = json.dumps({
        "source": source, "deps": dep_interfaces,
        "flags": flags, "compiler": compiler_version,
    }, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()[:12]

f1 = fingerprint("fn main() {}", {"core": "v3"}, ["-O2"], "rustc 1.85")
f2 = fingerprint("fn main() {}", {"core": "v4"}, ["-O2"], "rustc 1.85")
print("未变指纹:", f1)
print("依赖变化指纹:", f2)
print("缓存是否命中:", f1 == f2)

def propagate_dirty(adj, initial_dirty):
    """初始脏集合沿依赖图向下游传播."""
    dirty = set(initial_dirty)
    worklist = list(initial_dirty)
    while worklist:
        v = worklist.pop()
        for u in adj.get(v, []):        # adj: v -> 依赖 v 的模块
            if u not in dirty:
                dirty.add(u)
                worklist.append(u)
    return dirty

# adj: 谁依赖谁 (被依赖 -> 依赖者)
adj = {"core": ["ui", "main"], "ui": ["main"], "main": []}
print("core 变化后需要重编:", sorted(propagate_dirty(adj, {"core"})))
print("ui 变化后需要重编:", sorted(propagate_dirty(adj, {"ui"})))

# 接口 vs 实现的失效差异
def should_recompile(iface_hash_old, iface_hash_new, impl_only_changed):
    if impl_only_changed:
        return "下游无需重编 (实现隔离)"
    return "需要重编" if iface_hash_old != iface_hash_new else "无需重编"

print(should_recompile("a1", "a1", impl_only_changed=True))
print(should_recompile("a1", "b9", impl_only_changed=False))

class Memoizer:
    """以 (query, 输入) 为键的备忘录, 跨请求复用结果."""
    def __init__(self):
        self.cache = {}
        self.calls = 0
    def run(self, key, compute):
        if key not in self.cache:
            self.cache[key] = compute()
            self.calls += 1
        return self.cache[key]

m = Memoizer()
type_of = lambda k: f"type:{k}"
for _ in range(10):
    m.run(("typeck", "foo"), lambda: type_of("foo"))
print("实际计算次数:", m.calls, "(10 次请求只算了 1 次)")

# 跨运行持久化: 磁盘上的 query 缓存
class PersistentMemo:
    def __init__(self):
        self.cache = {}
    def get(self, key, deps_fingerprint):
        entry = self.cache.get(key)
        if entry and entry["fingerprint"] == deps_fingerprint:
            return entry["value"], True    # 缓存命中
        return None, False

pm = PersistentMemo()
pm.cache["typeck:foo"] = {"fingerprint": "abc", "value": "int"}
print(pm.get("typeck:foo", deps_fingerprint="abc"))   # 命中
print(pm.get("typeck:foo", deps_fingerprint="xyz"))   # 失效

import hashlib

class ArtifactCache:
    def __init__(self, store=None):
        self.store = store or {}
    def key(self, source, deps, flags):
        return hashlib.sha256((source + deps + flags).encode()).hexdigest()[:16]
    def get(self, source, deps, flags):
        k = self.key(source, deps, flags)
        if k in self.store:
            return self.store[k], "cache hit"
        artifact = f"obj:{k}"
        self.store[k] = artifact
        return artifact, "cache miss (已回填)"

ac = ArtifactCache()
for _ in range(3):
    obj, status = ac.get("src(main)", "core@v3", "-O2")
    print(f"  {status} -> {obj}")