#11 · Track A · 前端与 Orchestrator · 书 p.54

P01 · OutputSink：线程 → asyncio 的批量唤醒

orchestrator 线程每秒产生上万条输出，前端 asyncio 循环要消费它们；每条都 call_soon_threadsafe 太贵，只唤醒一次又容易丢唤醒。

⏱ 45 min 难度中实现 mine/p01_output_sink.py 验收 ./check.sh p01 提示 A-11 · 参考解 C-11 solutions/p01_output_sink.py 标记完成

题面速览

在系统中的位置：API 进程里有两个事件循环：前端循环（HTTP、SSE、tokenize）和 orchestrator 线程里的循环。跨线程唤醒 asyncio 只能用 loop.call_soon_threadsafe：它会往 loop 的 self-pipe 写一个字节并加锁，单次约微秒级，每秒上万次就很可观。
目标设计：

生产者（任意线程）                         消费者（前端 loop）
put(x):                                   get_batch():
  items.append(x)                           ① out = drain(); if out: return out
  if 已有唤醒在途: return                    ② event.clear(); 标记“无唤醒在途”   ← 武装
  标记“唤醒在途”                             ③ out = drain(); if out: return out  ← 关键
  call_soon_threadsafe(event.set)           ④ await event.wait()（带超时）

丢唤醒是怎么发生的：没有第 ③ 步时：消费者在 ① 发现队列空 → 生产者此刻 append 一条，看到“唤醒在途”（上一轮留下的）于是不唤醒 → 消费者 ② 武装 → ④ 睡下。那条消息一直躺在队列里，直到下一条 put 或超时。测试通过 after_empty_drain 钩子把一次 put 精确插进 ① 和 ② 之间。
需求：put 线程安全、永不阻塞、同一生产者顺序保持 · 一个“唤醒周期”内最多一次 call_soon_threadsafe，次数记在 wakeups_scheduled · get_batch 返回 1..max_items 条，超时返回 []。

验收要点：基本 FIFO · 前端未消费时 1000 次 put 只调度 1 次唤醒 · 超时返回 [] · 4 生产者 × 5000 条各自顺序不乱、消费者不饿死 · 乒乓 300 轮不丢唤醒 · “空 drain 之后、重新武装之前”注入的 put 必须立即被拿到。

直觉与类比

信箱、门铃和门上的小旗

你（前端 loop）住楼上，楼下有信箱（队列）。邮递员（orchestrator 线程）只能按门铃（call_soon_threadsafe）叫你下楼——每按一次你都要跑一趟楼梯，很贵。于是信箱旁插一面小旗（“唤醒在途”）：① 邮递员投信后，小旗倒着就竖起来并按一次铃；已竖着说明铃按过了、你正在下楼的路上，就不再按。② 你下楼把信一次拿光。③ 只有准备回楼上休息时才把小旗放倒。

丢唤醒的那一瞬间

你下楼打开信箱，空的，转身准备上楼。就在你还没放倒小旗的瞬间，邮递员投了一封信——他看小旗还竖着（上一轮留下的），以为铃按过了，没按就走了。你放倒小旗，上楼睡觉，那封信在信箱里躺到天亮。解法：放倒小旗之后再看一眼信箱（第二次 drain）。

框架视角 · T · SBWC（先自己想）

	问题	提示
T	生产者、消费者在哪？跨哪条边界？	多个普通线程 → 一个 asyncio loop；只能走 `call_soon_threadsafe` 这座桥
S	共享状态？谁读谁写？	“待取的数据”和“唤醒是否在途”。deque 的 append/popleft 在 CPython 里原子；标志的“检查并设置”不是，要锁
B	谁会阻塞？谁绝不能阻塞？	put 永不阻塞；get_batch 只能在 loop 里 await，不能 time.sleep 或用线程锁去等
F	不变量	“队列非空 ⇒ 要么消费者醒着，要么有一次唤醒在途”。上面的时间线在哪一步违反了它？
E	边界	超时；max_items 比积压少（剩下的下次 get_batch 会不会白等？）；多生产者交错

易错点

只 drain 一次就睡：例 2 场景卡到超时；乒乓式交互时不时卡住。
每次 put 都 call_soon_threadsafe：功能正确，1000 次 put 产生 1000 次唤醒。
先看标志再 append：生产者看到 False → 调度唤醒 → 消费者醒来 drain 为空 → 武装、睡下 → 生产者这时才 append 且不再唤醒，数据卡住。
用 asyncio.Queue 跨线程 put：不是线程安全的，偶发数据丢失或 loop 不醒；在生产者线程直接 event.set() 同理。
持锁调用钩子或持锁 await：死锁。
从不 event.clear()：Event 一直 set，wait 立即返回，空队列时在不让出的循环里空转（有 timeout 就烧满 CPU，None 就永远卡死）。
每圈重新计算完整 timeout：被伪唤醒多次后总等待时长超标；应入口算一次 deadline。

逐步图：例 2 的交错时间线

▶ 在“空 drain 之后、武装之前”插入一次 put

1. t0：上一轮没武装

上一轮 get_batch 直接从队列拿走 'a' 返回（没武装）。

标志 = True，队列 = ·，wakeups_scheduled = 1

2. t1：消费者第一次 drain

drain → []。标志 True，队列空。

3. t2：钩子里另一线程 put('b')

生产者 append('b')；看到标志 True → 不调度唤醒。

标志 True，队列 = b

4. t3：消费者武装

event.clear()；在锁里标志置 False。队列仍是 b

5. t4：第二次 drain → 拿到 'b'

_drain() → ['b']，立即返回（几乎 0 秒）。wakeups_scheduled 仍是 1。

返回后标志 False、Event clear：下一次 put 会正常调度唤醒，计数变成 2。

6. t4'：朴素实现会怎样

没有第二次 drain：直接 await event.wait() → 没人会 set（'b' 的 put 没调度唤醒）→ 睡满 timeout。

把“第二次 drain”那两行删掉再跑例 2，你会看到它卡满 2 秒——这是理解本题最快的办法。

三级提示

L1I/O 示例与关键问题I/O 示例 + 关键问题 + 细节（轻度剧透）

例 1：单线程里观察唤醒计数（s = OutputSink(); s.bind_loop(); s.put("a"); s.put("b"); s.put("c")）：

调用	返回	wakeups_scheduled	说明
3 次 put	—	1	第一次 put 发起唤醒；后两次看到“在途”，不再调度
`await s.get_batch(max_items=2)`	['a','b']	1	队列非空，立刻返回，不睡
`await s.get_batch(max_items=2)`	['c']	1	同上
`s.put("d")`	—	1	上一轮唤醒还“在途”（消费者没重新武装），不调度
`await s.get_batch()`	['d']	1	第一次 drain 就拿到
`await s.get_batch(timeout=0.05)`	[]	1	约 0.05s 后超时；这次真正“武装”了
`s.put("e")`	—	2	武装之后的第一次 put 才再调度一次
`await s.get_batch()`	['e']	2

例 3：消费者已在等（create_task(s.get_batch(timeout=2)) 后 sleep 0.01），另一线程连 put 5 条 → await task 得 [0,1,2,3,4]，wakeups_scheduled == 1。

关键问题

“唤醒在途”标志由谁置 True、谁置 False？直接从队列拿到数据返回时会被置回吗？有问题吗？
为什么 put() 不能直接 event.set()？asyncio.Event 和 call_soon_threadsafe 各自对线程有什么假设？
put() 里“append 数据”和“检查/设置标志”谁先谁后？反过来会出现什么交错？
“发现队列为空”和“武装”之间有一个窗口，put 落在这里且看到标志仍是 True，会怎样？怎样才能不错过？
event.clear() 什么时候调用？从不 clear 或在 drain 之前 clear 会怎样？
timeout 是整个 get_batch 的总时长，循环可能转好几圈，怎么保证不超？
钩子里另一个线程调用 put() 且钩子会等它结束；调用钩子时手里拿着 put 也要用的锁会怎样？

细节

唤醒周期：从“标志 False→True（某次 put 调度了唤醒）”到“消费者重新武装（置回 False）”。一个周期内 call_soon_threadsafe 最多一次；wakeups_scheduled 就是周期数。
loop 已关闭时 call_soon_threadsafe 抛 RuntimeError，生产者可以忽略。
put 可以拿一把很短的 threading.Lock（只保护标志的检查并设置），但不能等消费者。
“1..max_items 条或 []”：有东西就至少返回 1 条且不超过 max_items，剩下留给下次；只有超时才返回空。
bind_loop：不传参数时用 asyncio.get_running_loop()，并在这里创建 Event。bind 之前的 put 只需把数据放进队列。
钩子严格在“一次 drain 发现为空”之后、“重新武装”之前调用；不持锁。

L2思路与常见坑思路拆分与常见坑（中度剧透）

状态：deque（数据）、threading.Lock + bool（“唤醒在途”）、loop、asyncio.Event、wakeups_scheduled。标志的检查并设置在锁里，计数在同一临界区加。
put 的顺序：先 append，再在锁里“若标志 False 则置 True 并计数”，出锁后（若需要）call_soon_threadsafe(event.set)。先 append 保证：任何人看到“标志已被置起”时，数据一定已在队列里。
get_batch 是一个循环：drain → 非空返回 → 调钩子 → 武装（clear Event、锁里置标志 False）→ 再 drain → 非空返回 → 等 Event（剩余超时）→ 回到开头。武装和第二次 drain 之间不能有 await。
为什么第二次 drain 能堵住窗口：并发 put 的“看标志”要么在武装之前——它看到 True、没调度，但它的 append 已经发生，第二次 drain 一定拿到；要么在之后——看到 False，会调度唤醒，wait 会醒。两种都不丢。
超时：入口 deadline = loop.time() + timeout，每次等待前算剩余；≤ 0 返回 []。asyncio.wait_for(event.wait(), remaining)，捕获 TimeoutError 后再 drain 最后一次。
伪唤醒：Event 可能被上一轮遗留的 set 置位，wait 立即返回、drain 为空——不是错误，循环再转一圈就会武装并睡下。

L3详解：答案详解：答案、逐步追踪、骨架提示（重度剧透）

关键问题答案

put 置 True，消费者武装时置 False。直接拿到数据返回时不置回——没问题：消费者下次进来总是先 drain，只有真正准备睡前才武装；这段时间的 put 不唤醒，而此时也不需要唤醒（例 1 第 4 行）。
asyncio.Event 只能在所属 loop 线程上操作；call_soon_threadsafe 是唯一线程安全地让 loop 执行回调并把它从 select 中叫醒的方式。
必须先 append 再看标志；反过来的交错见“易错点”。
那次 put 不会唤醒你；唯一的补救是武装之后再 drain 一次（时间线 t4）。
clear 放在“武装”里，和“标志置 False”挨着，之后紧跟第二次 drain 和 await，中间无让出点。call_soon_threadsafe 排的回调只有在你 await 时才执行，所以 clear 和置 False 谁先谁后不影响正确性。从不 clear：wait 立即返回、空转（见易错点）。
入口算一次 deadline，每圈用 deadline - loop.time()。
死锁：钩子里的线程 put 要拿同一把锁，钩子在等那个线程结束。

面试官追问 & 对照源码

如果消费者有多个（多个 asyncio loop）还能这样做吗？
janus.Queue 帮你解决了什么、没解决什么？

对照源码：engine/async_engine_utils.py:40-97 · engine/omni_engine_base.py:973-991