σμimplied vol · KKmax(S−K,0)udβρᵢⱼfat tailsPnL · drawdownyield · TdSₜ = μSₜ dt + σSₜ dWₜC = S·N(d₁) − K·e⁻ʳᵀ·N(d₂)σₚ² = wᵀΣwd₁ = [ln(S/K) + (r + σ²⁄2)T] ⁄ σ√T∂V/∂t + ½σ²S²·∂²V/∂S² + rS·∂V/∂S − rV = 0Sharpe = (Rₚ − R_f) ⁄ σₚVaR₉₉ = μ − 2.33σρ = Cov(X,Y) ⁄ σₓσᵧE[Rₚ] = Σ wᵢ·E[Rᵢ]φ(x) = e^(−x²/2) ⁄ √2πE[Rᵢ] = R_f + βᵢ(E[Rₘ] − R_f)dXₜ = θ(μ − Xₜ)dt + σ dWₜσ²ₜ = ω + α·ε²ₜ₋₁ + β·σ²ₜ₋₁C − P = S − K·e⁻ʳᵀΔ = ∂V/∂S Γ = ∂²V/∂S²p = (e^(rΔt) − d) ⁄ (u − d)rₜ = ln(Pₜ ⁄ Pₜ₋₁)E[Xₜ₊₁ | ℱₜ] = Xₜf* = (bp − q) ⁄ bz = (x − μ) ⁄ σP(N=k) = λᵏe^(−λ) ⁄ k!Θ = −∂V/∂Tν = S·φ(d₁)·√Tmin wᵀΣw s.t. wᵀμ = μ*∫Σ∂∞λ
uantCraft Academy路线图学习路线
Ctrl K
AO
+1.2k 位学习者
登录加入 ↗
路线图学习路线
AO
+1.2k 位学习者
登录
  • English
  • Français
  • Español
  • العربية
  • Português
  • Deutsch
  • 日本語
  • 中文
σμimplied vol · KKmax(S−K,0)udβρᵢⱼfat tailsPnL · drawdownyield · TdSₜ = μSₜ dt + σSₜ dWₜC = S·N(d₁) − K·e⁻ʳᵀ·N(d₂)σₚ² = wᵀΣwd₁ = [ln(S/K) + (r + σ²⁄2)T] ⁄ σ√T∂V/∂t + ½σ²S²·∂²V/∂S² + rS·∂V/∂S − rV = 0Sharpe = (Rₚ − R_f) ⁄ σₚVaR₉₉ = μ − 2.33σρ = Cov(X,Y) ⁄ σₓσᵧE[Rₚ] = Σ wᵢ·E[Rᵢ]φ(x) = e^(−x²/2) ⁄ √2πE[Rᵢ] = R_f + βᵢ(E[Rₘ] − R_f)dXₜ = θ(μ − Xₜ)dt + σ dWₜσ²ₜ = ω + α·ε²ₜ₋₁ + β·σ²ₜ₋₁C − P = S − K·e⁻ʳᵀΔ = ∂V/∂S Γ = ∂²V/∂S²p = (e^(rΔt) − d) ⁄ (u − d)rₜ = ln(Pₜ ⁄ Pₜ₋₁)E[Xₜ₊₁ | ℱₜ] = Xₜf* = (bp − q) ⁄ bz = (x − μ) ⁄ σP(N=k) = λᵏe^(−λ) ⁄ k!Θ = −∂V/∂Tν = S·φ(d₁)·√Tmin wᵀΣw s.t. wᵀμ = μ*∫Σ∂∞λ
uantCraft Academy路线图学习路线
Ctrl K
AO
+1.2k 位学习者
登录加入 ↗
路线图学习路线
AO
+1.2k 位学习者
登录
  • English
  • Français
  • Español
  • العربية
  • Português
  • Deutsch
  • 日本語
  • 中文
Lock-free 学习路线›模块 3 · 无锁队列›主题 3.3·~1小时 10分钟高级

多生产者多消费者队列

路线进度模块 3/4 · 主题 3/3
3.3 多生产者多消费者队列
—实验得分— / 100总 XP—连续天数—

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

  1. 01票号不等于交付
  2. 02检查所有调度
  3. 03序号带来了什么
  4. 04你要实现什么

一个生产者加一个消费者时,每个位置只有一个写者。人多了,位置必须原子地分配,而且光有票号还不够:持有位置 5 的生产者还没把元素写进去。每个槽位上的序号说明轮到谁。在两个生产者的 2,756 种调度和两个消费者的 1,600 种调度上,这个队列从不出错。

01

票号不等于交付

有 capacity 个槽位的有界队列用两个只增不减的计数器。生产者用对 enq 的比较并交换领取下一个位置,消费者用对 deq 的比较并交换领取下一个位置。位置 p 对应槽位 p % capacity。比较并交换让领取变得安全:两个生产者绝不会拿到同一个位置。但领到位置不等于写入了元素。在交换与存入之间,槽位已被占用却还是空的,此时去读的消费者读到的是垃圾。所以每个槽位带一个只在正确时刻才前进的序号。

main.pyPython
def init_queue(capacity):    shared = {"enq": 0, "deq": 0}    for cell in range(capacity):        shared["seq%d" % cell] = cell    return shareddef enqueue(shared, item, capacity):    while True:        yield        pos = shared["enq"]        cell = pos % capacity        diff = shared["seq%d" % cell] - pos        if diff < 0:            return False        if diff > 0:            continue        yield        if compare_and_swap(shared, "enq", pos, pos + 1):            break    yield    shared["data%d" % cell] = item    shared["seq%d" % cell] = pos + 1    shared["put%d" % pos] = item    return Truedef dequeue(shared, capacity):    while True:        yield        pos = shared["deq"]        cell = pos % capacity        diff = shared["seq%d" % cell] - (pos + 1)        if diff < 0:            return None        if diff > 0:            continue        yield        if compare_and_swap(shared, "deq", pos, pos + 1):            break    yield    item = shared.get("data%d" % cell)    shared["seq%d" % cell] = pos + capacity    shared["got%d" % pos] = item    return itemsolo = lambda shared: [producer_thread((7,), 2)(shared), consumer_thread(1, 2)(shared)]for prefix in ([0, 0, 0], [0, 0, 0, 1, 1, 1], [0, 0, 1, 1, 0, 1, 1]):    print(prefix, run_prefix(solo, prefix, init_queue(2))[0])
[0, 0, 0] {'enq': 1, 'deq': 0, 'seq0': 1, 'seq1': 1, 'data0': 7, 'put0': 7}
[0, 0, 0, 1, 1, 1] {'enq': 1, 'deq': 1, 'seq0': 2, 'seq1': 1, 'data0': 7, 'put0': 7, 'got0': 7}
[0, 0, 1, 1, 0, 1, 1] {'enq': 1, 'deq': 0, 'seq0': 1, 'seq1': 1, 'data0': 7, 'put0': 7}

看这三次运行。[0, 0, 0] 之后,生产者已领取位置 0(enq: 1),存入 data0,并把 seq0 升到 1,元素就绪。消费者的三个回合之后,got0: 7、seq0: 2:该槽位为下一圈空了出来。最后一次运行里,消费者恰好在生产者领取与发布之间去看,看到 seq0 = 0 而不是 1,就空手而归。它没有读到垃圾,只是说“还没准备好”,元素会被之后的某次尝试取到。

02

检查所有调度

队列现在必须承诺什么?生产者把每个元素记在它的位置下(put3),消费者把取到的元素记在自己领取的位置下(got3)。队列正确的话,每个被取出的位置里放的就是当初放进去的元素(没有 mismatch),并且被取出的位置恰是 0、1、2……,没有空洞也没有重复(没有 gap)。这就给了队列它的顺序:位置的顺序。violations 从终态里读出这两个事实,verify 对任意数量的生产者和消费者运行探索器。这里是一个推两个元素的生产者对两个消费者,以及一个元素对一个消费者。

main.pyPython
def violations(final):    puts = {int(key[3:]): value for key, value in final.items() if key.startswith("put")}    gots = {int(key[3:]): value for key, value in final.items() if key.startswith("got")}    broken = []    if any(gots[p] != puts.get(p) for p in gots):        broken.append("mismatch")    if sorted(gots) != list(range(len(gots))):        broken.append("gap")    return sorted(broken)def verify(producers, consumers, capacity):    def make_threads(shared):        return [producer_thread(items, capacity)(shared) for items in producers] + [consumer_thread(n, capacity)(shared) for n in consumers]    outcomes = explore_all(make_threads, init_queue(capacity))    broken = {}    pushed = popped = 0    for key, count in outcomes.items():        final = dict(key)        pushed = max(pushed, sum(1 for k in final if k.startswith("put")))        popped = max(popped, sum(1 for k in final if k.startswith("got")))        for name in violations(final):            broken[name] = broken.get(name, 0) + count    return {"schedules": sum(outcomes.values()), "broken": dict(sorted(broken.items())), "pushed": pushed, "popped": popped}print(verify([(7,)], [1], 2))print(verify([(7, 8)], [1, 1], 2))
{'schedules': 4, 'broken': {}, 'pushed': 1, 'popped': 1}
{'schedules': 1600, 'broken': {}, 'pushed': 2, 'popped': 2}
定义序号

每个槽位上记录“轮到谁”的计数器。它只在元素被写入时(变为 p + 1)和被取走时(变为 p + capacity)才变化,所以读到它的线程不用任何锁就知道元素是否可以安全使用。

03

序号带来了什么

想知道它的价值,就把它拿掉。下面这个粗心的消费者用比较并交换领取位置,然后不看 seq 立刻读取槽位。面对只推一个元素的生产者,它有 20 种调度,其中 10 种在生产者还没发布任何东西时就领取了位置,并把 None 当作元素返回,或返回错误的元素。真正的队列对同一个程序只有 4 种调度,没有一种出错。序号恰恰就是把“已领取的票”变成“已交付的元素”的那道检查。

main.pyPython
def careless_dequeue(shared, capacity):    yield    pos = shared["deq"]    cell = pos % capacity    yield    if not compare_and_swap(shared, "deq", pos, pos + 1):        return None    yield    item = shared.get("data%d" % cell)    shared["got%d" % pos] = item    return itemdef careless_consumer(attempts, capacity):    def thread(shared):        for _ in range(attempts):            yield from careless_dequeue(shared, capacity)    return threadmake = lambda shared: [producer_thread((7,), 2)(shared), careless_consumer(1, 2)(shared)]outcomes = explore_all(make, init_queue(2))broken = sum(n for key, n in outcomes.items() if violations(dict(key)))print(sum(outcomes.values()), "schedules,", broken, "break the invariants")
20 schedules, 10 break the invariants
容量至少为 2

只有一个槽位时,表示“位置 p 已就绪”的值 p + 1,同时也是表示“为位置 p + 1 空出”的值,生产者就可能覆盖一个还没被读的元素。所以真实的实现把容量向上取整为 2 的幂(至少 2),这也让 % capacity 变成了廉价的掩码运算。

main.pyPython
ticks = [int(round(float(c) * 100)) for c in load_bars()["close"].iloc[:2]]print(ticks)print("two producers:", verify([(ticks[0],), (ticks[1],)], [2], 2))print("two consumers:", verify([tuple(ticks)], [1, 1], 2))
[9369, 9305]
two producers: {'schedules': 2756, 'broken': {}, 'pushed': 2, 'popped': 2}
two consumers: {'schedules': 1600, 'broken': {}, 'pushed': 2, 'popped': 2}
真实数据怎么说把 SPY 的前两个收盘价(9,369 和 9,305 美分)送过只有两个槽位的队列
两个生产者,一个消费者2,756种调度,0 种破坏不变量
一个生产者,两个消费者1,600种调度,0 种破坏不变量
粗心的消费者1020 种调度中读到不存在的元素

探索器作证:队列的两种形态在所有调度下都是正确的,而去掉序号检查的版本在它(小得多的)空间里有一半会出错。生产者和消费者从不互相等待:消费者发现没有就绪的元素就直接返回,生产者发现队列已满就返回 False。这就是无锁队列的代价,也是它的礼物:操作可能失败,但绝不会阻塞。

04

你要实现什么

实验沿着本讲走:带序号的空队列、通过比较并交换领取并发布的 enqueue 与 dequeue、对两条不变量的检查、可对任意数量生产者和消费者运行探索器的验证函数,以及用两个 SPY 收盘价做的研究。

  • 函数 · 问题依赖
  • init_queue(capacity)空队列是什么样子?简单算术
  • enqueue(shared, item, capacity)生产者如何领取并发布?简单算术
  • dequeue(shared, capacity)消费者如何领取并释放?简单算术
  • violations(final)终态破坏了哪些不变量?简单算术
  • verify(producers, consumers, capacity)队列在所有调度下都正确吗?explore_allviolations
  • study(bars)两种形态在 SPY 收盘价上都成立吗?verify

要点

  • 用比较并交换领票,用序号表示就绪。 领到位置它就是你的了;序号则告诉所有人其中的元素何时就绪。
  • 未就绪不是错误。 没找到就绪元素的消费者返回 None,稍后再试;它从不等待,也不读垃圾。
  • 用去掉序号的版本作对照来证明。 1,600 和 2,756 种调度没有违例;粗心的消费者在 20 种里有 10 种出错。
快速检查
一个队列有 4 个槽位。一个消费者领取了位置 6 并释放了它的槽位。该槽位的序号现在是多少,它为哪个位置空了出来?
请选择一个答案。

下一个主题:你已经靠跑遍所有调度证明了小程序的正确性。测试并发代码把这种做法与真实团队使用的工具作对比:一种是竞态检测器,它观察一次运行并标出可能发生竞争的访问;另一种是模型检验器,它做的事和你的探索器一样,但带有能让它扩展规模的技巧。

实验:评分内容

6 个任务 · 100 分
  • 空队列5 分
  • 入队25 分
  • 出队25 分
  • 不变量10 分
  • 在所有调度上验证15 分
  • 基于 SPY 表的研究20 分
资源管理器 · lab-lf-08
▾ lab-lf-08
▾ data
…
bash · 浏览器内
learner@quantcraft:~/lab-lf-08$
Python 尚未加载PythonLn 1, Col 1Spaces: 4UTF-8

要重置此主题吗?

lab-lf-08 中的所有内容将恢复为初始文件,终端也会被清空。你当前的解答将丢失,且无法撤销。

将被重置
  • main.py → 初始代码
  • 终端历史与测试输出
  • 未提交的更改
将被保留
  • 实验最高得分
  • 你的 XP 与连续天数
  • 讲解进度与笔记