内存模型：acquire 与 release

阅读约 5 分钟4 个代码示例

处理器可以按任意顺序执行两个互相独立的读和写，而另一个内核可能看到这种选择的结果。先发布一个 SPY 价格再举起标志，看到标志的读者仍可能读到价格 0。写者和读者的四种内存序组合里，只有一种是安全的：写者用 release，读者用 acquire，且二者配对。

处理器会重排

存储缓冲区、缓存和编译器都允许内核在单个线程察觉不到的前提下，先完成较晚的写或读，再完成较早的。但别的线程能察觉到。我们的机器直接对此建模：线程是一串操作，只要某个操作的顺序约束得到满足，就可以按任意顺序执行它。同一地址上的操作始终保持顺序，否则线程就看不到自己写的值。acquire加载会挡住它后面的一切；release存储则要等它前面的一切。两个函数说明了全部：must_wait 列出一个操作要等什么，eligible 列出接下来可以执行什么。

main.pyPython

def must_wait(ops, j):    waits = []    for i in range(j):        same = ops[i][1] == ops[j][1]        after_acquire = ops[i][0] == "load" and ops[i][3] == "acquire"        release = ops[j][0] == "store" and ops[j][3] == "release"        if same or after_acquire or release:            waits.append(i)    return waitsdef eligible(ops, done):    return [j for j in range(len(ops)) if j not in done and all(i in done for i in must_wait(ops, j))]writer = [("store", "data", 1, "relaxed"), ("store", "flag", 1, "release")]reader = [("load", "flag", "r1", "acquire"), ("load", "data", "r2", "relaxed")]print(must_wait(writer, 1), eligible(writer, set()), eligible(writer, {0}))print(must_wait(reader, 1), eligible(reader, set()))print(eligible([("load", "flag", "r1", "relaxed"), ("load", "data", "r2", "relaxed")], set()))

[0] [0] [1]
[0] [0]
[0, 1]

标志的 release 存储要等位置 0 的数据存储，所以写者只能按顺序执行操作。标志的 acquire 加载挡住了它后面的数据加载，所以读者也只能按顺序执行。若用普通加载，两个加载同时可执行，读者就可能先读数据。

运行所有顺序

outcomes 是前几个主题的探索器在这台新机器上的版本。状态包括每个线程已完成的操作、内存和寄存器。从一个状态出发，它尝试每个线程的每个可执行操作，当没有操作可执行时，寄存器就构成一种结果。因为许多顺序会到达同一个状态，它会记住见过的状态。消息传递测试用两次存储发布（先数据后标志），用两次加载消费（先标志后数据）。消费者看到 r1 == 1（标志）而 r2 == 0（数据还没到）时就错了，这由 reads_stale 检查。

main.pyPython

def outcomes(programs, memory=None):    found = set()    seen = set()    def visit(done, mem, regs):        state = (tuple(frozenset(d) for d in done), tuple(sorted(mem.items())), tuple(sorted(regs.items())))        if state in seen:            return        seen.add(state)        moved = False        for tid, ops in enumerate(programs):            for j in eligible(ops, done[tid]):                moved = True                kind, address, x, _ = ops[j]                mem2, regs2 = dict(mem), dict(regs)                if kind == "store":                    mem2[address] = x                else:                    regs2[x] = mem.get(address, 0)                done2 = [set(d) for d in done]                done2[tid].add(j)                visit(done2, mem2, regs2)        if not moved:            found.add(tuple(sorted(regs.items())))    visit([set() for _ in programs], dict(memory or {}), {})    return founddef mp_programs(write_order, read_order):    return [        [("store", "data", 1, "relaxed"), ("store", "flag", 1, write_order)],        [("load", "flag", "r1", read_order), ("load", "data", "r2", "relaxed")],    ]def reads_stale(results):    return any(dict(outcome).get("r1") == 1 and dict(outcome).get("r2") == 0 for outcome in results)for pair in (("relaxed", "relaxed"), ("release", "acquire")):    results = outcomes(mp_programs(*pair))    print(pair, sorted(results), "stale:", reads_stale(results))

('relaxed', 'relaxed') [(('r1', 0), ('r2', 0)), (('r1', 0), ('r2', 1)), (('r1', 1), ('r2', 0)), (('r1', 1), ('r2', 1))] stale: True
('release', 'acquire') [(('r1', 0), ('r2', 0)), (('r1', 0), ('r2', 1)), (('r1', 1), ('r2', 1))] stale: False

用 relaxed 内存序时，消费者可能得到全部四组 (r1, r2)，包括过期的 (1, 0)。给标志的存储加 release、给标志的加载加 acquire 后只剩三种：看到标志的读者也一定看到数据，因为双方都不能跨过标志被重排。

release 与 acquire 要配对

只有一边够吗？把四种组合都试一遍。只在写者加 release 不行：读者可能先于标志加载数据。只在读者加 acquire 也不行：写者可能先于数据存储标志。只有配对才行，下表展示了这一点。这就是 C++ std::atomic 和 Rust 的 release-acquire约定，也是无锁队列用 release 存储下标来发布一个槽位、再用 acquire 加载来读取它的原因。

main.pyPython

PAIRS = [("relaxed", "relaxed"), ("release", "relaxed"), ("relaxed", "acquire"), ("release", "acquire")]def ordering_table():    return {pair: reads_stale(outcomes(mp_programs(*pair))) for pair in PAIRS}for pair, stale in ordering_table().items():    print("write", pair[0], "| read", pair[1], "->", "can read stale data" if stale else "safe")

write relaxed | read relaxed -> can read stale data
write release | read relaxed -> can read stale data
write relaxed | read acquire -> can read stale data
write release | read acquire -> safe

acquire 与 release 是一对，不是咒语

没有同一变量上与之对应的 acquire，release 什么也保护不了。C++ 里真正的内存序如 seq_cst 更强也更贵；relaxed 最便宜，但不保证任何顺序。选能让上面的测试通过的最弱的内存序，并用探索器来证明。

main.pyPython

price = int(round(float(load_bars()["close"].iloc[0]) * 100))programs = mp_programs("release", "acquire")programs[0][0] = ("store", "data", price, "relaxed")print(price, sorted(outcomes(programs)))

9369 [(('r1', 0), ('r2', 0)), (('r1', 0), ('r2', 9369)), (('r1', 1), ('r2', 9369))]

真实数据怎么说把 SPY 第一个收盘价以 9,369 美分通过标志发布

relaxed + relaxed4种结果，可能读到过期价格

仅 release4种结果，可能读到过期价格

仅 acquire4种结果，可能读到过期价格

release + acquire3种结果，绝不过期

以 SPY 的第一个收盘价 9,369 美分为例，在四种配置中的三种里，读者可能看到标志的同时读到价格 0。这类缺陷在恰好保持顺序的机器上能通过所有测试，换一台机器就会暴露。所以必须像探索器那样，对所有允许的顺序证明那一对是安全的。

定义release-acquire 配对

对一个标志的 release 存储，以及读取它的 acquire 加载。写者在 release 之前做的一切，读者在 acquire 之后都能看到。它是让“通过标志发布数据”正确的、代价最小的内存序。

你要实现什么

实验沿着本讲走：一个操作必须等待什么、接下来能执行哪些操作、一组线程的所有结果、消息传递测试、对过期数据的检查、四种内存序的表，以及发布真实 SPY 价格的研究。

函数 · 问题依赖
must_wait(ops, j)这个操作必须等待什么？简单算术
eligible(ops, done)接下来可以执行哪些操作？简单算术
outcomes(programs, memory)线程最终可能得到什么？eligible
mp_programs(write_order, read_order)消息传递长什么样？简单算术
reads_stale(results)读者能不能看到标志却没有数据？简单算术
ordering_table()哪些内存序组合是安全的？outcomesmp_programs
study(bars)哪些组合能保护真实的 SPY 价格？outcomesmp_programs

快速检查

写者先存数据，再对标志做 release 存储。读者先对标志做普通（relaxed）加载，再对数据做普通加载。读者能否看到标志已置位而数据缺失？

请选择一个答案。

下一个主题：你现在有了构建无锁结构的工具，也有了描述它保证什么的词汇。进展保证会给 obstruction-free、lock-free 和 wait-free 这三个级别命名，并拿你写过的每个计数器去对付一个在最糟时刻被永远暂停的线程。

实验：评分内容

7 个任务 · 100 分

操作要等待什么15 分
接下来能执行哪些10 分
每种结果25 分
消息传递测试10 分
过期数据5 分
四种内存序10 分
基于 SPY 表的研究25 分

def must_wait(ops, j): waits = [] for i in range(j): same = ops[i][1] == ops[j][1] after_acquire = ops[i][0] == "load" and ops[i][3] == "acquire" release = ops[j][0] == "store" and ops[j][3] == "release" if same or after_acquire or release: waits.append(i) return waitsdef eligible(ops, done): return [j for j in range(len(ops)) if j not in done and all(i in done for i in must_wait(ops, j))]writer = [("store", "data", 1, "relaxed"), ("store", "flag", 1, "release")]reader = [("load", "flag", "r1", "acquire"), ("load", "data", "r2", "relaxed")]print(must_wait(writer, 1), eligible(writer, set()), eligible(writer, {0}))print(must_wait(reader, 1), eligible(reader, set()))print(eligible([("load", "flag", "r1", "relaxed"), ("load", "data", "r2", "relaxed")], set()))

def outcomes(programs, memory=None): found = set() seen = set() def visit(done, mem, regs): state = (tuple(frozenset(d) for d in done), tuple(sorted(mem.items())), tuple(sorted(regs.items()))) if state in seen: return seen.add(state) moved = False for tid, ops in enumerate(programs): for j in eligible(ops, done[tid]): moved = True kind, address, x, _ = ops[j] mem2, regs2 = dict(mem), dict(regs) if kind == "store": mem2[address] = x else: regs2[x] = mem.get(address, 0) done2 = [set(d) for d in done] done2[tid].add(j) visit(done2, mem2, regs2) if not moved: found.add(tuple(sorted(regs.items()))) visit([set() for _ in programs], dict(memory or {}), {}) return founddef mp_programs(write_order, read_order): return [ [("store", "data", 1, "relaxed"), ("store", "flag", 1, write_order)], [("load", "flag", "r1", read_order), ("load", "data", "r2", "relaxed")], ]def reads_stale(results): return any(dict(outcome).get("r1") == 1 and dict(outcome).get("r2") == 0 for outcome in results)for pair in (("relaxed", "relaxed"), ("release", "acquire")): results = outcomes(mp_programs(*pair)) print(pair, sorted(results), "stale:", reads_stale(results))

('relaxed', 'relaxed') [(('r1', 0), ('r2', 0)), (('r1', 0), ('r2', 1)), (('r1', 1), ('r2', 0)), (('r1', 1), ('r2', 1))] stale: True ('release', 'acquire') [(('r1', 0), ('r2', 0)), (('r1', 0), ('r2', 1)), (('r1', 1), ('r2', 1))] stale: False

PAIRS = [("relaxed", "relaxed"), ("release", "relaxed"), ("relaxed", "acquire"), ("release", "acquire")]def ordering_table(): return {pair: reads_stale(outcomes(mp_programs(*pair))) for pair in PAIRS}for pair, stale in ordering_table().items(): print("write", pair[0], "| read", pair[1], "->", "can read stale data" if stale else "safe")