「垃圾回收算法」

import sys

class Obj:
    def __init__(self, name):
        self.name = name
        self.refs = []

def mark_from_roots(roots, heap):
    """朴素可达性标记: 从根集做图遍历."""
    marked, stack = set(), list(roots)
    while stack:
        o = stack.pop()
        if id(o) in marked:
            continue
        marked.add(id(o))
        stack.extend(heap.get(id(o), []))    # 对象的出边
    return marked

a, b = Obj("a"), Obj("b")
heap = {id(a): [b], id(b): []}
alive = mark_from_roots([a], heap)
print("存活:", [o.name for o in (a, b) if id(o) in alive])

# 引用计数是另一种判定方式: 计数归零即可回收
class RCObj:
    def __init__(self, name):
        self.name, self.rc = name, 0
    def add_ref(self):
        self.rc += 1
    def release(self):
        self.rc -= 1
        return self.rc == 0            # True 表示可立即回收

x = RCObj("x")
x.add_ref()
print("释放后是否可回收:", x.release())

# 标记-清除: 标记存活, 清扫未标记的
def mark_sweep(heap, roots, allocated):
    marked = mark_from_roots(roots, heap)
    freed = [oid for oid in allocated if oid not in marked]
    for oid in freed:
        allocated.remove(oid)         # 回收, 但留下的空洞难以利用
    return freed

# 标记-复制: 存活对象搬到另一半, 原空间整体清空
def mark_copy(heap, roots, allocated, to_space):
    marked = mark_from_roots(roots, heap)
    mapping = {}
    for i, oid in enumerate(sorted(marked)):
        mapping[oid] = ("to", i)       # 新地址: 紧凑排列
    return mapping                     # 无碎片, 但要预留一半空间

# 标记-整理: 存活对象向一端滑动, 保持顺序
def mark_compact(heap, roots, allocated):
    marked = mark_from_roots(roots, heap)
    order = sorted(marked)             # 按原地址排序, 保持相对顺序
    return {oid: i for i, oid in enumerate(order)}

def cheney(from_space, roots):
    """Cheney 复制: scan 指针遍历新空间, free 指针分配."""
    to_space, forward = [], {}
    free = 0
    def copy(oid):
        nonlocal free
        if oid in forward: return forward[oid]
        to_space.append(oid)
        forward[oid] = free
        free += 1
        return forward[oid]
    queue = [copy(r) for r in roots]
    scan = 0
    while scan < len(to_space):
        oid = to_space[scan]
        scan += 1
        for child in from_space.get(oid, []):
            queue.append(copy(child))
    return to_space

print(cheney({1: [2], 2: [3], 3: []}, [1]))

import random

def simulate_lifetimes(n=100_000, max_age=20):
    """模拟对象寿命分布: 大多数很快死亡, 少数长寿."""
    ages = []
    for _ in range(n):
        if random.random() < 0.9:            # 90% 的对象很短命
            ages.append(random.randint(1, 2))
        else:
            ages.append(random.randint(3, max_age))
    young = sum(1 for a in ages if a <= 2)
    return young / n, ages

ratio, ages = simulate_lifetimes()
print(f"短命对象占比: {ratio:.2%}")

class GenerationalHeap:
    def __init__(self, promote_age=15, eden_cap=100):
        self.eden, self.survivor, self.old = [], [], []
        self.age, self.promote_age, self.eden_cap = {}, promote_age, eden_cap

    def allocate(self, obj):
        self.eden.append(obj)
        if len(self.eden) > self.eden_cap:
            self.minor_gc()

    def minor_gc(self):
        survivors = [o for o in self.eden if self.is_alive(o)]
        for o in survivors:
            self.age[o] = self.age.get(o, 0) + 1
            if self.age[o] >= self.promote_age:
                self.old.append(o)             # 晋升
            else:
                self.survivor.append(o)
        self.eden = []

    def is_alive(self, obj):
        return getattr(obj, "alive", False)

WHITE, GRAY, BLACK = 0, 1, 2

def tri_color_mark(roots, children):
    color = {}
    def get(o): return color.get(o, WHITE)
    gray = [r for r in roots]
    for r in gray: color[r] = GRAY
    while gray:
        o = gray.pop()
        for c in children(o):
            if get(c) == WHITE:
                color[c] = GRAY
                gray.append(c)
        color[o] = BLACK
    return color

children = lambda o: {1: [2, 3], 2: [3], 3: []}.get(o, [])
print(tri_color_mark([1], children))

# 增量更新 (incremental update): 记录「黑->白」的新引用, 事后重新扫描黑对象
def incremental_update_barrier(black_obj, new_ref, dirty_set):
    dirty_set.add(black_obj)          # 把黑对象重新变灰
    return new_ref

# 快照-at-the-beginning (SATB): 记录被覆盖的旧引用, 保证按标记开始时的快照不丢对象
def satb_barrier(old_ref, buffer):
    if old_ref is not None:
        buffer.append(old_ref)        # 保存旧值, 后续作为根重扫
    return old_ref

dirty, buf = set(), []
incremental_update_barrier("A", "C", dirty)
satb_barrier("B", buf)
print("增量更新脏集:", dirty, "| SATB 缓冲:", buf)

// 写屏障在生成代码中的位置: 每次引用字段赋值都要经过它
void set_field(Obj *owner, Obj **field, Obj *value) {
    write_barrier(owner, field, value);   // 记录卡表或 SATB 缓冲
    *field = value;
}

def concurrent_cycle(roots, children, user_mutations):
    """并发标记: 用户线程改引用时把黑对象重新入脏集重扫."""
    color = tri_color_mark(roots, children)
    dirty = {owner for owner, ref in user_mutations
             if color.get(owner) == BLACK and color.get(ref, WHITE) == WHITE}
    return tri_color_mark(list(dirty), children) if dirty else color

def g1_select_regions(regions, pause_budget):
    """G1 的选择策略: 按垃圾比例排序, 在预算内尽量多回收."""
    ordered = sorted(regions, key=lambda r: r["garbage_ratio"], reverse=True)
    chosen, cost = [], 0
    for r in ordered:
        if cost + r["cost"] > pause_budget:
            break
        chosen.append(r["id"])
        cost += r["cost"]
    return chosen, cost

regions = [{"id": i, "garbage_ratio": 0.9 - i * 0.1, "cost": 1} for i in range(5)]
print(g1_select_regions(regions, pause_budget=3))

# 常用诊断: 先观察, 再调参
java -Xlog:gc*:file=gc.log:time,uptime,level,tags -jar app.jar
# 关键指标: 各代回收频率、暂停时长分布、晋升速率、堆使用曲线

# 延迟优先: 选低延迟收集器并给暂停目标
java -XX:+UseZGC -XX:MaxGCPauseMillis=10 -Xmx8g -jar app.jar
java -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -Xmx8g -jar app.jar

# 吞吐优先: 加大堆、减少回收次数
java -XX:+UseParallelGC -Xms16g -Xmx16g -jar app.jar

# 用弱引用做缓存: 让 GC 在内存紧张时自动清理
import weakref

class Cache:
    def __init__(self):
        self._store = weakref.WeakValueDictionary()
    def get(self, k):
        return self._store.get(k)
    def put(self, k, v):
        self._store[k] = v            # 无强引用时自动消失, 不会撑爆堆

c = Cache()
print(c.get("missing"))