进展保证：从 obstruction-free 到 wait-free

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

三个计数器在所有调度下都给出正确答案，但它们并不一样。在最糟的时刻暂停一个线程：自旋锁会让所有线程停住，比较并交换计数器还能让某个线程通过，而 fetch-and-add 计数器让每个线程都在 1 个回合内完成。这就是进展保证阶梯的三级。

一个被暂停的线程

真实的操作系统可以随时把线程从内核上拿下来：时间片用完、缺页把它送去磁盘、来了一个中断。好的并发结构应当扛得住这种情况。我们用一个对手把问题说精确：它先执行自己选择的一段回合前缀 prefix，然后把某个线程永远暂停，再给其他所有线程 budget 轮机会。如果仍有没被暂停的线程没完成，这个结构就能被停住。自旋锁以最简单的方式没通过这个测试：对手暂停持锁的线程，其余线程就一直空转到预算用完。

main.pyPython

def locked_increments(k):    def thread(shared):        for _ in range(k):            while True:                yield                if not shared.get("lock", 0):                    shared["lock"] = 1                    break            yield            value = shared.get("x", 0)            yield            shared["x"] = value + 1            yield            shared["lock"] = 0    return threaddef progress_without(make_threads, prefix, paused, budget=50):    n = len(make_threads({}))    allowed = [tid for tid in range(n) if tid not in paused]    _, alive = run_prefix(make_threads, list(prefix) + allowed * budget)    return [tid for tid in allowed if tid not in alive]lk = lambda shared: [locked_increments(1)(shared), locked_increments(1)(shared)]print(run_prefix(lk, [0]))print("thread 0 paused holding the lock:", progress_without(lk, [0], [0]))print("thread 0 finished first:         ", progress_without(lk, [0, 0, 0, 0], [0]))

({'lock': 1}, [0, 1])
thread 0 paused holding the lock: []
thread 0 finished first:          [1]

一个回合之后线程 0 持有了锁（'lock': 1）。在这时暂停它，线程 1 在五十轮里什么也完成不了：[]。如果先让线程 0 跑完它的四个回合，锁就重新空闲，线程 1 便能完成：[1]。区别只在于对手选择在哪里下手。锁把临界区内的任何一次暂停，变成了整个程序的暂停。

无锁：总有人能完成

is_blocking 把对手自动化了。它尝试所有几个回合的前缀，依次暂停每个存活的线程，并问 progress_without 其余线程是否还能完成。对自旋锁，只需一个回合答案就是“能被阻塞”。对比较并交换计数器则永远不会：一个线程在读和交换之间被暂停，只是让值保持不变，所以另一个线程的交换会成功。这就是无锁的定义：整个系统总在前进，哪怕有一个线程被永远停住。

main.pyPython

def is_blocking(make_threads, depth=3, budget=50):    def visit(prefix):        _, alive = run_prefix(make_threads, prefix)        for paused in alive:            finished = progress_without(make_threads, prefix, [paused], budget)            if len(finished) < len([tid for tid in range(len(make_threads({}))) if tid != paused]):                return True        if len(prefix) < depth:            return any(visit(prefix + [tid]) for tid in alive)        return False    return visit([])cs = lambda shared: [cas_increments(1)(shared), cas_increments(1)(shared)]at = lambda shared: [atomic_increments(1)(shared), atomic_increments(1)(shared)]print("spin lock:", is_blocking(lk))print("compare-and-swap:", is_blocking(cs))print("fetch-and-add:", is_blocking(at))

spin lock: True
compare-and-swap: False
fetch-and-add: False

比较并交换计数器和 fetch-and-add 计数器都是非阻塞的。阶梯有第三级，是因为“无锁”仍允许某个线程倒霉：它可能输掉每一场竞争。

无等待：人人都能完成

无等待的操作，其自身回合数有一个任何邻居都无法打破的上界。worst_own_turns 就是在测它：一个线程在所有完整调度中所占回合数的最大值。然后 classify 提出关键的问题：邻居有 1 次增量和有 3 次增量时，线程 0 的最坏情形是否相同？如果相同，线程 0 就是 wait-free；如果增长，邻居就能让线程 0 多重试几次，这个结构只是 lock-free。

main.pyPython

def worst_own_turns(make_threads, tid):    worst = 0    def visit(prefix):        nonlocal worst        _, alive = run_prefix(make_threads, prefix)        if not alive:            worst = max(worst, prefix.count(tid))            return        for other in alive:            visit(prefix + [other])    visit([])    return worstdef classify(factory, others=(1, 3)):    def program(m):        return lambda shared: [factory(1)(shared), factory(m)(shared)]    if is_blocking(program(others[0])):        return "blocking"    light, busy = worst_own_turns(program(others[0]), 0), worst_own_turns(program(others[1]), 0)    return "wait-free" if light == busy else "lock-free"for neighbour in (1, 2, 3):    cas = lambda shared: [cas_increments(1)(shared), cas_increments(neighbour)(shared)]    atomic = lambda shared: [atomic_increments(1)(shared), atomic_increments(neighbour)(shared)]    print(neighbour, "increments next door: CAS", worst_own_turns(cas, 0), "turns, fetch-and-add", worst_own_turns(atomic, 0))print([classify(f) for f in (locked_increments, cas_increments, atomic_increments)])

1 increments next door: CAS 4 turns, fetch-and-add 1
2 increments next door: CAS 6 turns, fetch-and-add 1
3 increments next door: CAS 8 turns, fetch-and-add 1
['blocking', 'lock-free', 'wait-free']

怎么选

wait-free 最强，设计起来也最贵，而 fetch-and-add 表明它并非总是遥不可及。lock-free 是队列和栈的实际目标。锁用在持锁者被暂停也无妨的地方没问题，但绝不能用在不允许停顿的路径上。

main.pyPython

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 (("locked", locked_increments), ("cas", cas_increments), ("atomic", atomic_increments)):    program = lambda shared, f=factory: [f(k)(shared) for k in ups]    blocking = is_blocking(program)    print(name, classify(factory), blocking, None if blocking else worst_own_turns(program, 0))

locked blocking True None
cas lock-free False 8
atomic wait-free False 1

真实数据怎么说SPY 前 6 个收益率的上涨日：线程 0 有 1 次增量，线程 1 有 3 次

自旋锁blocking一个线程被暂停就让另一个停住

比较并交换lock-free线程 0 最坏情形：8 个回合

fetch-and-addwait-free线程 0 最坏情形：1 个回合

按邻居工作量统计线程 0 的最坏自身回合数

fetch-and-add，邻居 3 次1

CAS，邻居 1 次4

CAS，邻居 2 次6

CAS，邻居 3 次8

看柱状图：不管邻居多忙，fetch-and-add 线程都只要 1 个回合，而比较并交换线程随着邻居增量变多需要 4、6、8 个回合。邻居每多一次增量，线程 0 就可能再输一场竞争、重复一次读和交换。这种增长恰恰是 lock-free 与 wait-free 的分界。

定义进展保证

阻塞：一个被暂停的线程就能让其他线程停住。无锁：总有某个线程能完成。无等待：每个线程都在有限个自己的回合内完成。在这个列表里，每一级都蕴含前一级。

你要实现什么

实验沿着本讲走：自旋锁计数器、在部分线程被暂停时运行、判断暂停一个线程是否会让其他线程停住的测试、线程自身回合数的最坏情形、对计数器的分类，以及对 SPY 上涨日的三种计数器的研究。

函数 · 问题依赖
locked_increments(k)等待锁的计数器？简单算术
progress_without(make_threads, prefix, paused, budget)有线程被暂停时谁能完成？run_prefix
is_blocking(make_threads, depth, budget)一个被暂停的线程能让所有人停住吗？progress_without
worst_own_turns(make_threads, tid)一个线程最多需要多少个回合？run_prefix
classify(factory, others)阻塞、无锁还是无等待？is_blockingworst_own_turns
study(bars)三个计数器在 SPY 上各处于哪一级？classifyis_blockingworst_own_turns

快速检查

一个线程用比较并交换做 1 次递增，另一个线程做 5 次递增。下列哪个说法是有保证的？

请选择一个答案。

下一个主题：你现在知道了队列可以承诺什么。单生产者单消费者环形缓冲区会构建最简单而实用的无锁队列：一个写者、一个读者、固定内存、完全不需要原子读-改-写，并用探索器检验它从不丢失或重排任何一个 tick。

实验：评分内容

6 个任务 · 100 分

自旋锁上的计数器10 分
在线程被暂停时运行20 分
一次暂停能让所有人停住吗20 分
线程的最坏情形15 分
给保证命名10 分
基于 SPY 表的研究25 分

def locked_increments(k): def thread(shared): for _ in range(k): while True: yield if not shared.get("lock", 0): shared["lock"] = 1 break yield value = shared.get("x", 0) yield shared["x"] = value + 1 yield shared["lock"] = 0 return threaddef progress_without(make_threads, prefix, paused, budget=50): n = len(make_threads({})) allowed = [tid for tid in range(n) if tid not in paused] _, alive = run_prefix(make_threads, list(prefix) + allowed * budget) return [tid for tid in allowed if tid not in alive]lk = lambda shared: [locked_increments(1)(shared), locked_increments(1)(shared)]print(run_prefix(lk, [0]))print("thread 0 paused holding the lock:", progress_without(lk, [0], [0]))print("thread 0 finished first: ", progress_without(lk, [0, 0, 0, 0], [0]))

def is_blocking(make_threads, depth=3, budget=50): def visit(prefix): _, alive = run_prefix(make_threads, prefix) for paused in alive: finished = progress_without(make_threads, prefix, [paused], budget) if len(finished) < len([tid for tid in range(len(make_threads({}))) if tid != paused]): return True if len(prefix) < depth: return any(visit(prefix + [tid]) for tid in alive) return False return visit([])cs = lambda shared: [cas_increments(1)(shared), cas_increments(1)(shared)]at = lambda shared: [atomic_increments(1)(shared), atomic_increments(1)(shared)]print("spin lock:", is_blocking(lk))print("compare-and-swap:", is_blocking(cs))print("fetch-and-add:", is_blocking(at))

def worst_own_turns(make_threads, tid): worst = 0 def visit(prefix): nonlocal worst _, alive = run_prefix(make_threads, prefix) if not alive: worst = max(worst, prefix.count(tid)) return for other in alive: visit(prefix + [other]) visit([]) return worstdef classify(factory, others=(1, 3)): def program(m): return lambda shared: [factory(1)(shared), factory(m)(shared)] if is_blocking(program(others[0])): return "blocking" light, busy = worst_own_turns(program(others[0]), 0), worst_own_turns(program(others[1]), 0) return "wait-free" if light == busy else "lock-free"for neighbour in (1, 2, 3): cas = lambda shared: [cas_increments(1)(shared), cas_increments(neighbour)(shared)] atomic = lambda shared: [atomic_increments(1)(shared), atomic_increments(neighbour)(shared)] print(neighbour, "increments next door: CAS", worst_own_turns(cas, 0), "turns, fetch-and-add", worst_own_turns(atomic, 0))print([classify(f) for f in (locked_increments, cas_increments, atomic_increments)])

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 (("locked", locked_increments), ("cas", cas_increments), ("atomic", atomic_increments)): program = lambda shared, f=factory: [f(k)(shared) for k in ups] blocking = is_blocking(program) print(name, classify(factory), blocking, None if blocking else worst_own_turns(program, 0))