丢失更新的原因是读和写是中间有间隙的两个回合。原子操作把间隙堵上：读、改、写在一个不可分割的步骤里完成。有了它，计数器在所有调度下都正确：调度从 28 种变成 4 种，而且没有一种出错。
一个回合，没有间隙
在我们的机器里，一个回合从一个 yield 走到下一个 yield，所以写在两个 yield 之间的内容不会被交错。因此不含 yield 的函数就表现为原子操作。最常见的是fetch-and-add：它读出旧值、写入 old + n、返回 old，全部在一个回合内。调用它的计数线程整个增量只有一个 yield 而不是两个，于是两个线程再也不会读到同一个旧值。
def fetch_add(shared, name, n): old = shared.get(name, 0) shared[name] = old + n return olddef atomic_increments(k): def thread(shared): for _ in range(k): yield fetch_add(shared, "x", 1) return threadtwo = lambda shared: [atomic_increments(1)(shared), atomic_increments(1)(shared)]for prefix in ([0, 1], [1, 0]): print(prefix, run_prefix(two, prefix))
[0, 1] ({'x': 2}, [])
[1, 0] ({'x': 2}, [])两种顺序都以 {'x': 2} 结束，没有线程剩下。每次增量只有一个回合，程序只有 [0, 1] 和 [1, 0] 两种调度（研究里增量为 1 和 3 时共 4 种），没有一种会丢更新。
比较并交换与重试循环
fetch-and-add 只会做加法。其他情形需要通用工具比较并交换（CAS）。其模式是一个循环：读出旧值，据此算出新值，请 CAS 仅在这期间没人改过该值时才写入。CAS 失败说明有人先动了，于是重新读取再试。下面是两个 CAS 计数器：线程 1 在线程 0 写入之前读取，它的 CAS 失败一次、记一次重试，第二次尝试成功。更新不会丢，因为过期的写入会被拒绝。
def compare_and_swap(shared, name, expected, new): if shared.get(name, 0) != expected: return False shared[name] = new return Truedef cas_increments(k): def thread(shared): for _ in range(k): while True: yield old = shared.get("x", 0) yield if compare_and_swap(shared, "x", old, old + 1): break fetch_add(shared, "retries", 1) return threadcas = lambda shared: [cas_increments(1)(shared), cas_increments(1)(shared)]print(run_prefix(cas, [0, 1, 0, 1]))print(run_prefix(cas, [0, 1, 0, 1, 1, 1]))
({'x': 1, 'retries': 1}, [1])
({'x': 2, 'retries': 1}, [])[0, 1, 0, 1] 之后共享字典是 {'x': 1, 'retries': 1}，线程 1 还活着：它读到 0，但线程 0 已经存入了 1，所以它的比较并交换被拒绝。再给线程 1 接下来的两个回合（重新读、重新交换），最终得到 x = 2。注意代价：输的一方把活干了两遍。fetch-and-add 从不重试；CAS 则随线程冲突的次数而重试。
回合数会变化时如何探索
上一个主题的探索器假设每个线程占用固定的回合数 lengths。重试循环打破了这一点：输掉竞争的线程会多占回合，多少取决于调度。解决办法是把调度当作一棵树来搜索。从一段回合前缀出发，run_prefix 重放程序并告诉我们哪些线程还活着；若一个都没有，该调度已完整，统计其最终状态；否则对每个活着的线程分叉。这种深度优先搜索恰好访问每个完整调度一次。随后 values_of 从全部结果中读出某个共享名字的取值。
def explore_all(make_threads, shared=None): outcomes = {} def visit(prefix): final, alive = run_prefix(make_threads, prefix, shared) if not alive: key = tuple(sorted(final.items())) outcomes[key] = outcomes.get(key, 0) + 1 return for tid in alive: visit(prefix + [tid]) visit([]) return outcomesdef values_of(outcomes, name): return sorted({dict(key).get(name, 0) for key in outcomes})print(explore_all(cas))racy = lambda shared: [increments(1)(shared), increments(1)(shared)]print(values_of(explore_all(racy), "x"), values_of(explore_all(cas), "x"))
{(('x', 2),): 2, (('retries', 1), ('x', 2)): 4}
[1, 2] [2]在这台机器里，每个操作都会被所有线程同时且按顺序地看到。真实的处理器对普通读写并不保证这一点，即便是原子操作也会让各个内核争抢缓存行。下一个主题会去掉第一个假设。
returns = [float(r) for r in load_bars()["close"].pct_change().dropna()][:6]ups = [sum(1 for r in returns[start::2] if r > 0) for start in (0, 1)]for name, factory in (("racy", increments), ("atomic", atomic_increments), ("cas", cas_increments)): outcomes = explore_all(lambda shared, f=factory: [f(k)(shared) for k in ups]) print(name, sum(outcomes.values()), "schedules, x in", values_of(outcomes, "x"))
racy 28 schedules, x in [1, 2, 3, 4] atomic 4 schedules, x in [4] cas 144 schedules, x in [4]
两种修复在所有调度下都正确，区别在于代价。fetch-and-add 每次增量一个回合就完成：4 种调度。CAS 计数器有 144 种调度，因为冲突会增加回合，最坏情况下一个线程重试 3 次。这就是无锁代码的权衡：没人在锁上等待，但每场竞争的赢家会迫使输家把活重做一遍。
把共享值读出、据此算出新值、再作为一个不可分割的步骤写回的操作。fetch-and-add 和比较并交换是最常用的两个，所有无锁结构都由它们搭成。
你要实现什么
实验沿着本讲走：fetch-and-add、比较并交换、基于它们各自的计数线程、能处理回合数可变的线程的探索器、从全部结果中读取某个名字取值的辅助函数，以及对 SPY 上涨日的三种计数器的研究。
- 函数 · 问题依赖
- fetch_add(shared, name, n)加 n，之前是什么值？简单算术
- compare_and_swap(shared, name, expected, new)值还是我读到的那个吗？简单算术
- atomic_increments(k)没有间隙的计数器？fetch_add
- cas_increments(k)用重试代替丢失的计数器？compare_and_swapfetch_add
- explore_all(make_threads, shared)无论回合数多少，所有结果是什么？run_prefix
- values_of(outcomes, name)这个名字取哪些值？简单算术
- study(bars)三种计数器在 SPY 上有何不同？explore_allvalues_of
下一个主题：我们一直假设每个线程立即且按顺序看到每一次写入。真实的处理器并不保证这一点。内存模型：acquire 与 release 会给机器加上存储缓冲区，演示读者在看到它所保护的数据之前就看到了标志，并用代价最小的内存序修复。
实验：评分内容
7 个任务 · 100 分- fetch-and-add5 分
- 比较并交换10 分
- 用 fetch-and-add 的计数器10 分
- 带重试循环的计数器20 分
- 探索所有调度20 分
- 某个名字的取值5 分
- 基于 SPY 表的研究30 分