P13 · ChunkAdapter：后台收发线程与顺序/清理语义

例 1 见逐步图（注意 key 里的数字：发送端用自己的 stage_id，接收端用 src_stage_id 拼同一个 key）。

例 2：快照隔离 + poll_ready 是“取走自上次以来的”

live = ["a"]; rcv.load_async("v")
snd.save_async("v", live, is_final=False)
live.append("b")                        # 调度线程继续改活对象
snd.save_async("v", live, is_final=True)
live.append("c")
# 收到的 data 依次是 ["a"]、["a", "b"]；conn.puts == ["v_0_0", "v_0_1"]

snd.save_async("q", 1, False); snd.save_async("q", 2, False)    # 稍等
rcv.poll_ready() -> {"q": [{"data": 1, ...}, {"data": 2, ...}]}
rcv.poll_ready() -> {}                                          # 已经被取走了
snd.save_async("q", 3, True)                                    # 稍等
rcv.poll_ready() -> {"q": [{"data": 3, "finished": True}]}

chunk 在 load_async 之前就已被 put，load_async 之后照样能收到。

例 3：cancel_sender（build_payload 阻塞在 gate 上）：save "p0"（开始 build，卡住）→ 排队 p1、p2、p3 → save "end"(final) → cancel_sender("w") → 放行。接收端得到 [{"data": "p0", "finished": False}, {"data": None, "finished": True}]，conn.puts == ["w_0_0", "w_0_1"]。p0 会不会被发出取决于取消时它是否已过了取消检查——两种都可接受；但 id 必须连续、final 必须到。

关键问题

1. save_async 在调度线程上执行，必须立即返回。它最多只能做哪两件事？build_payload 为什么不能在这里调用？
2. chunk_id 什么时候 +1：入队时、build 之后、还是 put 成功之后？
3. build_payload 返回 None 时什么情况跳过、什么情况仍要 put？final 块被 cancel 后 data 是什么？
4. 接收端怎么知道下一个要 get 的 key？每个请求要记住哪些状态？收到 finished 后还要轮询吗？
5. recv 线程在锁外 get 时，调度线程 cleanup 又 load_async 同一个 req。recv 回来怎么判断“这块属于已作废的那次注册”？
6. 10 个请求都在轮询但都没数据时 recv 线程该做什么？完全没有请求时呢？
7. stop() 要迅速返回。save 线程正睡在 Condition.wait() 上，怎么叫醒？如果它恰好在 notify 之前一瞬间才去 wait 呢？

细节

• save_async / cancel_sender / load_async / poll_ready / cleanup_receiver 都在调度线程上调用；SAVE 线程只做 build + put；RECV 线程只做 get + 提交。
• put 的对象固定是 {"data": payload, "finished": is_final}；poll_ready 原样返回，按 id 顺序放在 {req: [...]}；没有新内容的请求不出现。
• per-registration token：load_async 时生成 object() 存起来；cleanup 时删掉；recv 线程 get 之前记下 token，回来后在锁内用 is 比较。
• Condition.wait(timeout=...) 返回后必须重新检查条件；给 wait 加较短超时（如 0.1s），丢通知也只是多睡一会儿。

• 发送端：FIFO deque (req, snapshot, is_final) + Condition 唤醒 save 线程；next_send_id: dict[req, int]；cancelled: set[req]。只有一个 save 线程按 FIFO 处理，同一请求的 chunk 天然有序。
• save 线程一次迭代：取一项 → 查 cancelled（短暂持锁）→ 被取消且非 final：丢弃 → 被取消且 final：data=None → 否则锁外 build → data 为 None 且非 final：跳过 → put 到当前 id 的 key → 成功后 id+1；final 时清理 id 计数和 cancelled 标记。put 抛异常时不 +1。
• 接收端：registered: dict[req, token]、next_recv_id、ready: dict[req, list]、轮询队列 + Condition。load_async 幂等。
• recv 线程一轮：对当前队列里的每个请求轮询一次：锁内读 (token, next_id)，没注册了就扔掉；锁外 get；没数据放回队尾；有数据锁内比对 token → 一致才 append、id+1；finished 就注销（不再放回），否则放回。一轮结束：队列空 → Condition 上睡（带超时）；有请求但没进展 → 睡 ~1ms；有进展 → 立刻下一轮。
• poll_ready 原子交换：锁内把 ready 整个换成新的空 dict 并返回旧的。
• cleanup_receiver 是提交栅栏：锁内删除 registered、next_recv_id、ready。由于“比对 token + 提交”也在同一把锁内，cleanup 返回之后任何提交要么已发生（已被删掉），要么会看到 token 不匹配而丢弃。
• stop()：设 Event，对两个 Condition 分别 with cond: cond.notify_all()，再 join（带超时）。线程设为 daemon。