P05 · RpcRouter：共享通道上的 RPC 关联

例 1：两个线程并发调用，结果乱序回来

r = RpcRouter()
def send_for(key, val, delay):     # 模拟 router 线程稍后回传结果
    return lambda: threading.Timer(delay, lambda: r.deliver(key, val)).start()

线程 1: r.call("sleep", send_for("sleep", "ok-sleep", 0.05), timeout=1)   -> "ok-sleep"
线程 2: r.call("wake",  send_for("wake",  "ok-wake",  0.01), timeout=1)   -> "ok-wake"
# 结果在 send() 内部就同步回来了，也不能丢：
r.call(("rpc", 7), lambda: r.deliver(("rpc", 7), {"gpu_mem": 3}), timeout=0.1)  -> {"gpu_mem": 3}

例 2：超时、迟到、key 复用、异常结果

r.call("ping", lambda: None, timeout=0.1)     # 约 0.1s 后抛 TimeoutError
r.deliver("ping", "pong")                     # -> False（已经没人在等）
r.late_results                                # -> 1
r.call("ping", lambda: r.deliver("ping", "pong2"), timeout=0.1)   # -> "pong2"
r.call("bad", lambda: r.deliver("bad", ValueError("no such method")), timeout=0.1)
# -> 抛出 ValueError：结果本身是异常对象时 raise 而不是 return
w = r.register("dup"); r.register("dup")      # 第二次 -> ValueError

例 3：fail_all 广播并锁存：3 个线程 call("w0"/"w1"/"w2", send=lambda: None, timeout=5) 阻塞；主线程 r.fail_all(RuntimeError("orch dead")) → 3 个线程几乎立刻各自抛出；之后 call("after", …) 立即抛、send 不会被调用；register("x") 同样立即抛；late_results 不变。

关键问题

1. 为什么必须先 register 再 send()？构造一个“先 send 后 register”会永远等下去的时序。
2. waiter 为什么是 Queue(maxsize=1)？deliver 放结果时会不会阻塞？
3. 一个 waiter 会被谁从表里移除：deliver、call 的 finally，还是两者？怎么保证不互相踩？
4. unregister(key, waiter) 为什么要多传一个 waiter，并且只在“表里那个就是我这个”时才删？
5. deliver 需要锁吗？锁里做什么、锁外做什么？
6. waiter.get(timeout=...) 超时抛什么异常？对外抛什么？
7. fail_all 之后，正在等的和以后才来的调用分别怎么处理？锁存存在哪、谁来检查？

细节

• key 是任意 Hashable（字符串或 ("rpc", 7) 元组）；同一时刻一个 key 只能有一个活着的 waiter。
• deliver：有 waiter 交给它返回 True；否则返回 False 且 late_results += 1。
• 结果通道上传回来的可能是异常对象（远端执行失败或 fail_all 的致命错误）：call 拿到 BaseException 实例就 raise。

• 状态：threading.Lock、waiters: dict[key, Queue]、fatal: BaseException | None、late_results。
• register：锁内：fatal 已设置 → raise fatal；key 已在表 → ValueError；否则新建 Queue(maxsize=1) 放进表并返回。
• deliver：锁内 pop(key, None)——pop 保证同一 key 只交付一次；拿不到就计数返回 False。锁外 put_nowait(result)。
• unregister：锁内仅当 waiters.get(key) is waiter 时删除。
• call：w = register(key) → try: send(); w.get(timeout)，queue.Empty 转成 TimeoutError；结果是异常就 raise，否则 return → finally: unregister(key, w)。register 在 try 外：register 失败时没有东西需要注销。
• fail_all：锁内设置 fatal、取出所有 waiter、清空表；锁外逐个 put_nowait(error)（防御性忽略 queue.Full）。