两个线程给同一个计数器加 1,计数器最后却是错的,既没有报错也没有崩溃。x += 1 不是一步而是三步,两个线程如何交错这些步骤,决定了结果。在 SPY 的 1,085 个上涨日里,有 542 次更新凭空消失;加上锁就能全部找回。
一行代码,三个步骤
处理器不能直接对内存里的值做加法。它先把该值载入寄存器,再对寄存器加,最后把寄存器存回内存。所以 x += 1 就是 load x、add 1、store x。单个线程把这三步连着做,不会出问题。有两个线程时,机器可以先执行其中一个的一步,再执行另一个的一步,顺序任意。为了看清这一点,我们写一台小机器。线程是一串步骤,每个线程有自己的寄存器;线程之间共享一个由名字索引的值字典。函数 execute 执行某个线程的一步。下面我们手工走完两个线程的六步,轮流各走一步。
def execute(op, shared, regs, locks, tid): kind = op[0] if kind == "load": regs[tid] = shared.get(op[1], 0) elif kind == "add": regs[tid] += op[1] elif kind == "store": shared[op[1]] = regs[tid] elif kind == "lock": if op[1] in locks: return False locks[op[1]] = tid elif kind == "unlock": del locks[op[1]] return Trueshared, regs, locks = {"x": 0}, {0: 0, 1: 0}, {}turns = [(0, 0), (1, 0), (0, 1), (1, 1), (0, 2), (1, 2)]1for tid, step in turns: execute(INCREMENT[step], shared, regs, locks, tid) print("thread", tid, INCREMENT[step], "->", shared, regs)
thread 0 ('load', 'x') -> {'x': 0} {0: 0, 1: 0}
thread 1 ('load', 'x') -> {'x': 0} {0: 0, 1: 0}
thread 0 ('add', 1) -> {'x': 0} {0: 1, 1: 0}
thread 1 ('add', 1) -> {'x': 0} {0: 1, 1: 1}
thread 0 ('store', 'x') -> {'x': 1} {0: 1, 1: 1}
thread 1 ('store', 'x') -> {'x': 1} {0: 1, 1: 1}- 1两个线程都在对方存回之前载入:各自的寄存器里都是 0。
逐行读输出。线程 0 载入 0,线程 1 也载入 0(线程 0 还没存回),两者都对值为 0 的寄存器加 1,然后都存回 1。两个线程各做了一次 x += 1,x 却是 1 而不是 2。丢了一次更新。任何一个线程的代码本身都没错,这个缺陷只存在于这种步骤顺序之中。
结果取决于线程如何交错其步骤的缺陷。同一个程序在某些调度下正确、在另一些下出错,因此测试时很容易漏掉。
调度就是一串线程编号
在真实世界里,调度由操作系统选择,每次运行都不同,所以竞态那么难复现。在我们的机器里,由我们自己选。调度就是一串线程编号,run 给每一项所指的线程一个回合。有两种简单的调度很重要。串行调度里,线程 0 先跑完所有步骤,再轮到线程 1:这是我们希望的情形。轮转调度里,线程轮流各走一步:这是我们这个计数器的最坏情形,因为两个线程的步骤处处重叠。函数 lost_updates 比较增量本该加上的量与计数器实际显示的值。
def run(programs, schedule, shared=None): shared = dict(shared or {}) regs = {tid: 0 for tid in range(len(programs))} pcs = [0] * len(programs) locks = {} for tid in schedule: if pcs[tid] < len(programs[tid]) and execute(programs[tid][pcs[tid]], shared, regs, locks, tid): pcs[tid] += 1 while any(pcs[t] < len(programs[t]) for t in range(len(programs))): progress = False for tid in range(len(programs)): if pcs[tid] < len(programs[tid]) and execute(programs[tid][pcs[tid]], shared, regs, locks, tid): pcs[tid] += 1 progress = True if not progress: raise RuntimeError("deadlock") return shareddef serial_schedule(programs): return [tid for tid, program in enumerate(programs) for _ in program]def round_robin_schedule(programs): longest = max((len(program) for program in programs), default=0) return [tid for step in range(longest) for tid, program in enumerate(programs) if step < len(program)]def lost_updates(programs, schedule, name="x"): expected = sum(op[1] for program in programs for op in program if op[0] == "add") return expected - run(programs, schedule).get(name, 0)two = [INCREMENT, INCREMENT]print(serial_schedule(two), run(two, serial_schedule(two)))print(round_robin_schedule(two), run(two, round_robin_schedule(two)))print("lost:", lost_updates(two, round_robin_schedule(two)))
[0, 0, 0, 1, 1, 1] {'x': 2}
[0, 1, 0, 1, 0, 1] {'x': 1}
lost: 1串行调度 [0, 0, 0, 1, 1, 1] 得到 2;轮转调度 [0, 1, 0, 1, 0, 1] 得到 1,丢了一次更新。注意调度提前结束时 run 的做法:未完成的线程继续按线程顺序每人走一步,所以每个程序最终总会跑完。还要注意,已完成的线程会忽略它的回合,因此任何线程编号列表都是合法的调度。
锁把几步变成一个整体
锁通过禁止重叠来修复竞态。线程在 load 之前拿锁,在 store 之后放锁。这期间如果第二个线程请求这把锁,是拿不到的:它的回合什么也不做,停在同一步,直到锁空出来。于是 lock 与 unlock 之间的步骤作为一个整体运行,别的线程无法插进来。函数 counter_program 构造一个递增计数器的线程程序,可以带锁也可以不带。锁带来了新的危险,代码末尾演示了:两个线程以相反的顺序拿两把锁,各自握着对方需要的那把,谁也动不了。这就是死锁,当完整地走一遍仍毫无进展时,我们的 run 会检测到它。
def counter_program(increments, locked=False): program = [] for _ in range(increments): program += ([("lock", "m")] if locked else []) + INCREMENT + ([("unlock", "m")] if locked else []) return programsafe = [counter_program(1, locked=True), counter_program(1, locked=True)]print(counter_program(1, locked=True))print(run(safe, round_robin_schedule(safe)))ab = [[("lock", "a"), ("lock", "b"), ("unlock", "b"), ("unlock", "a")], [("lock", "b"), ("lock", "a"), ("unlock", "a"), ("unlock", "b")]]try: run(ab, [0, 1])except RuntimeError as error: print("error:", error)
[('lock', 'm'), ('load', 'x'), ('add', 1), ('store', 'x'), ('unlock', 'm')]
{'x': 2}
error: deadlock等待锁的线程什么也不做,而持锁的线程可能在任何一步被操作系统暂停。在交易热路径上,这意味着无法设定上界的抖动。本技能的其余部分将构建完全不需要锁的结构。
def study(bars): returns = [float(r) for r in bars["close"].pct_change().dropna()][:2000] ups = [sum(1 for r in returns[start::2] if r > 0) for start in (0, 1)] free = [counter_program(k) for k in ups] locked = [counter_program(k, locked=True) for k in ups] return {"up": sum(ups), "serial": run(free, serial_schedule(free))["x"], "racy": run(free, round_robin_schedule(free))["x"], "locked": run(locked, round_robin_schedule(locked))["x"]}print(study(load_bars()))
{'up': 1085, 'serial': 1085, 'racy': 543, 'locked': 1085}没有锁时,轮转调度让两个线程始终相差一步,因此它们总是载入同一个值,每一对增量只算一次:1,085 个上涨日大约丢掉一半。加锁能恢复准确的总数,代价是让一个线程等待。
你要实现什么
实验沿着本讲走:线程的一步、在调度下运行整台机器、两种简单调度、丢失更新的计数、带锁和不带锁的计数线程程序,以及用两个线程数 SPY 上涨日的研究。
- 函数 · 问题依赖
- execute(op, shared, regs, locks, tid)一步做什么?线程需要等待吗?简单算术
- run(programs, schedule, shared)在这个调度下线程会留下什么?execute
- serial_schedule(programs)一个线程接一个线程?简单算术
- round_robin_schedule(programs)轮流各走一步?简单算术
- lost_updates(programs, schedule)有多少次增量消失?run
- counter_program(increments, locked)计数线程做什么?简单算术
- study(bars)竞态在 SPY 上会丢多少次更新?counter_programrunlost_updates
下一个主题:你是靠手工选一种调度才找到那次丢失的更新的。真实程序的调度多到无法手工尝试。在交错探索器里,你要枚举一个小程序的全部调度,数出让它出错的那些,并把这个探索器作为本技能其余部分的工具。
实验:评分内容
7 个任务 · 100 分- 线程的一步15 分
- 调度下的机器25 分
- 一个线程接一个线程5 分
- 轮流各走一步10 分
- 丢失的更新10 分
- 计数线程10 分
- 基于 SPY 表的研究25 分