执行摘要

• 一句话：NIXL 推送注册读 _remote_agents 加锁，消除竞态误判
• 推荐动作：值得精读，尤其是把「小于 10 行的源码改动 + 100 行并发测试」这种配比作为规范参考。关注三个设计决策：(1) 为什么是「锁内快照 + 锁外发送」而不是「整个 check-then-send 持锁」；(2) 为什么只把 dict pop 放进临界区、把 NIXL teardown 留在锁外；(3) 两条回归测试如何用 threading.Event + 超时断言把「被锁挡住」与「功能正确」区分开——这一点在评审并发补丁时经常被做成只验证行为的假测试。审查时建议额外确认：基类 _handshake_lock 是 RLock 还是 Lock、以及快照后发送到已被移除 agent 的 NIXL 语义。

功能与动机

PR body 明确说明这是 NixlPushMode 可靠性清单（#48633）中的 X1 项：_do_send_reg_notif 运行在 nixl-push-writer 线程上，对 _remote_agents 做 check-then-send，但「Nothing guards that read, even though the lock comment in base_worker.py says _handshake_lock 'protects _handshake_futures and _remote_agents'」。两个线程会改这个 map：handshake done-callback 在锁内插入，TTL 淘汰 _evict_stale_engines → _cleanup_remote_engine 弹出。writer 可能先用 _ensure_handshake 确认引擎在线，随后条目被并发淘汰摘走，导致「the registration is treated as failed and _handle_failed_transfer fires on a request whose handshake succeeded moments earlier」，在 D 侧频繁 churn（rollout、autoscaling）时表现为偶发且不确定的 No remote agents for engine ... 报错。

实现拆解

1. 定位竞争窗口（读侧）：vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py 的 _do_send_reg_notif 原先直接 self._remote_agents.get(engine_id) 拿到活引用后立刻遍历发送，读与写完全无同步。
2. push_worker.py 改为锁内快照：在 with self._handshake_lock: 内执行 agents = dict(self._remote_agents.get(engine_id) or {})。这样既让 check-then-send 的「check」与 map 的所有写入互斥，又顺手消除了持活引用遍历时潜在的 dictionary changed size during iteration；dict() 拷贝只发生在临界区内，send_notif 等慢路径（NIXL 调用）仍留在锁外，锁竞争开销可以忽略。
3. base_worker.py 补齐写侧对称性：_cleanup_remote_engine 把 self._remote_agents.pop(engine_id) 包进同一把 _handshake_lock，并先把 agents 取出，再在锁外逐个 remove_remote_agent，即注释所述「Pop under the handshake lock; NIXL teardown stays outside it」。这样淘汰不再能从 writer 的读下方抽走条目，_remote_agents 真正满足锁注释声明的不变式；_evict_stale_engines 的调用点不变，仍是主线程经 _ensure_handshake 触发。
4. 测试配套：tests/v1/kv_connector/unit/test_nixl_push_connector.py 中 _StubWriterWorker.fresh() 新增 w._handshake_lock = threading.RLock() 以复刻基类字段，并新增 TestPushWriterRegSend（test_sends_to_all_agents_with_prefix_and_payload 校验通知前缀与 msgpack 载荷、test_fails_request_when_engine_not_handshaked 校验真未握手时仍走 _handle_failed_transfer、test_agent_read_is_serialized_with_handshake_lock 用「人为占锁 + threading.Event 超时」证明 writer 被挡在临界区外）与模块级 test_cleanup_remote_engine_pops_under_handshake_lock（证明淘汰 pop 同样被锁挡住，且放锁后 NIXL agent 恰好被移除一次）。
5. 无配置/接口/部署改动：未触碰 NixlPushConnector 的构造参数、通知协议或 metadata 结构，属于纯内部同步语义修复，对其它 connector（MooncakeStore 等）与通用 NixlConnectorWorker 行为无影响。

关键文件：

• vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py（模块 推送写入器；类别 source；类型 core-logic；符号 _do_send_reg_notif, _on_handshake）: 修复的主入口：_do_send_reg_notif 在 _handshake_lock 下快照 _remote_agents，消除 check-then-send 与 handshake 插入/淘汰的竞争。
• vllm/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py（模块 连接器基类；类别 source；类型 core-logic；符号 _cleanup_remote_engine, _evict_stale_engines）: 写侧对称修复：_cleanup_remote_engine 把 _remote_agents.pop 移入 _handshake_lock，使淘汰不再能打断 writer 的读。
• tests/v1/kv_connector/unit/test_nixl_push_connector.py（模块 推送单元测试；类别 test；类型 test-coverage；符号 TestPushWriterRegSend, _reg_send_worker, test_sends_to_all_agents_with_prefix_and_payload, test_fails_request_when_engine_not_handshaked）: 新增 TestPushWriterRegSend 与两条并发回归用例，并给 stub 补上 _handshake_lock，直接钉住本次修复的同步语义。

关键符号：_do_send_reg_notif, _cleanup_remote_engine, _evict_stale_engines, _ensure_handshake, TestPushWriterRegSend.test_agent_read_is_serialized_with_handshake_lock, test_cleanup_remote_engine_pops_under_handshake_lock

关键源码片段

vllm/distributed/kv_transfer/kv_connector/v1/nixl/push_worker.py

修复的主入口：_do_send_reg_notif 在 _handshake_lock 下快照 _remote_agents，消除 check-then-send 与 handshake 插入/淘汰的竞争。

def _do_send_reg_notif(self, req_id: str, reg_data: dict[str, Any]) -> None:
    """在 nixl-push-writer 线程上把 PUSH_REG 通知发给目标引擎的所有 rank。"""
    engine_id = reg_data["remote_engine_id"]
    notif_msg = PUSH_REG_NOTIF_PREFIX + msgspec.msgpack.encode(reg_data)
​
    # _remote_agents 会被其它线程改写：handshake done-callback 在锁内插入，
    # TTL 淘汰 _cleanup_remote_engine 在锁内 pop。必须用同一把 _handshake_lock
    # 做快照，否则 check-then-send 之间条目可能被淘汰线程摘走，
    # 导致 handshake 明明成功的请求被误判失败（X1）。
    with self._handshake_lock:
        agents = dict(self._remote_agents.get(engine_id) or {})
​
    if not agents:
        logger.error(
            "No remote agents for engine %s; cannot send registration for %s",
            engine_id,
            req_id,
        )
        # 只有确实没有已完成 handshake 的 agent 才判定失败
        self._handle_failed_transfer(req_id, None)
        return
​
    # 发送留在锁外：send_notif 是 NIXL 慢路径，不应拉长持锁时间
    for rank, agent_name in agents.items():
        try:
            self.nixl_wrapper.send_notif(agent_name, notif_msg=notif_msg)
        except Exception as e:
            self._log_failure(
                failure_type="push_reg_notif_failed",
                req_id=req_id,
                error=e,
                remote_rank=rank,
            )
    logger.debug(
        "Sent PUSH_REG for %s to engine %s (%dB)", req_id, engine_id, len(notif_msg)
    )

vllm/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py

写侧对称修复：_cleanup_remote_engine 把 _remote_agents.pop 移入 _handshake_lock，使淘汰不再能打断 writer 的读。

def _cleanup_remote_engine(
    self, engine_id: EngineId, *, log_eviction: bool = True
) -> None:
    """清理单个远端引擎的全部状态，TTL 淘汰与 shutdown 共用。
​
    调用方：_evict_stale_engines（主线程，经 _ensure_handshake 触发）。
    """
    assert engine_id in self._remote_agents
​
    # Notif-only 引擎（push 模式 D 侧）没有 descriptor 状态，可安全 pop
    for handle in self.dst_xfer_side_handles.pop(engine_id, {}).values():
        self.nixl_wrapper.release_dlist_handle(handle)
​
    # 关键修复（X1）：pop 必须与 push writer 的读共用 _handshake_lock，
    # 否则淘汰可以在 writer 完成 check 之后、send 之前摘掉条目，
    # 让一个刚刚 handshake 成功的请求被判定为注册失败。
    # 只把 pop 放进临界区，NIXL 资源释放留在锁外以免拉长持锁时间。
    with self._handshake_lock:
        agents = self._remote_agents.pop(engine_id)
    for agent_name in agents.values():
        self.nixl_wrapper.remove_remote_agent(agent_name)
​
    # 以下 per-engine 字典均由引擎主线程访问，无需加锁
    self.kv_caches_base_addr.pop(engine_id, None)
    self.dst_num_blocks.pop(engine_id, None)
    self.dst_region_num_blocks.pop(engine_id, None)
    self.dst_region_group_ids.pop(engine_id, None)
    self.dst_uses_region_group_mapping.pop(engine_id, None)
    self.dst_region_mem_types.pop(engine_id, None)
    self.tp_mappings.pop(engine_id, None)
    if self.transfer_topo is not None:
        self.transfer_topo.unregister_remote_engine(engine_id)
​
    # 丢弃缓存的时钟偏移，下次 handshake 重新测量
    self._engine_clock_offset.pop(engine_id, None)
    # 刚完成的 handshake 可能还没记录活跃时间，容忍缺失
    last_active = self._engine_last_active.pop(engine_id, None)
    if log_eviction and last_active is not None:
        logger.info(
            "Evicted stale remote engine %s (inactive for %.1fs).",
            engine_id,
            time.perf_counter() - last_active,
        )

tests/v1/kv_connector/unit/test_nixl_push_connector.py

新增 TestPushWriterRegSend 与两条并发回归用例，并给 stub 补上 _handshake_lock，直接钉住本次修复的同步语义。

class TestPushWriterRegSend:
    """_do_send_reg_notif：PUSH_REG 要发给目标 P 引擎所有已 handshake 的 agent，
    且 agent map 的读取必须在 _handshake_lock 下完成，
    以免与 done-callback 的插入或并发淘汰（X1）竞争。"""
​
    def test_agent_read_is_serialized_with_handshake_lock(self):
        """另一线程持锁（模拟 handshake done-callback 写 _remote_agents）时，
        writer 的 check-then-send 必须被挡在临界区之外。"""
        w, failures = self._reg_send_worker()
        w._remote_agents["prefill-engine"] = {(0, 0): "agent-0"}
        rd = _registration_data("req-1")
​
        w._handshake_lock.acquire() # 人为占住锁
        done = threading.Event()
        errors: list[BaseException] = []
​
        def _send():
            try:
                w._do_send_reg_notif("req-1", rd)
            except BaseException as e: # pragma: no cover
                errors.append(e)
            finally:
                done.set()
​
        t = threading.Thread(target=_send)
        t.start()
        # 锁未释放前：既不能发送，也不能误判失败
        assert not done.wait(timeout=0.2)
        w.nixl_wrapper.send_notif.assert_not_called()
        assert failures == []
​
        w._handshake_lock.release()
        # 放锁后 writer 恢复，恰好成功发送一次
        assert done.wait(timeout=2.0)
        t.join(timeout=2.0)
        assert errors == []
        assert failures == []
        assert w.nixl_wrapper.send_notif.call_count == 1
​
​
def test_cleanup_remote_engine_pops_under_handshake_lock():
    """回归（X1）：淘汰不得在其它线程处于临界区时摘掉 _remote_agents 条目，
    例如 push writer 正在读该 map 以发送注册通知。"""
    w = _eviction_worker(engine_ttl=30.0)
    w._remote_agents["D-old"] = {(0, 0): "agent-D-old"}
​
    w._handshake_lock.acquire()
    done = threading.Event()
​
    def _cleanup():
        w._cleanup_remote_engine("D-old")
        done.set()
​
    t = threading.Thread(target=_cleanup)
    t.start()
    assert "D-old" in w._remote_agents # 被锁挡住，条目仍在
​
    w._handshake_lock.release()
    assert done.wait(timeout=2.0)
    t.join(timeout=2.0)
    assert "D-old" not in w._remote_agents
    w.nixl_wrapper.remove_remote_agent.assert_called_once_with("agent-D-old")

评论区精华

本 PR 的 review 讨论极少：claude[bot] 只留了一条流程性评论（fork PR 自动 review 被禁用），维护者 DarkLight1337 直接 APPROVED，review_comments 为 0——因此技术论证几乎全部来自 PR body 与 commit message。PR body 里的两个设计要点值得记录：(1)「the writer can pass _ensure_handshake with the engine present, then lose it to a concurrent eviction before .get() runs」，明确了失败发生在 check 与 send 之间，而不是 handshake 本身失败；(2)「NIXL teardown stays outside it」，说明作者有意缩小临界区，只让 dict 操作进锁。issue 评论只有作者两次触发 CI（/ci run、/ci retry），第二次重试了 Buildkite CI #88172 中 3 个失败 job，未针对失败原因展开讨论。

• 锁内快照 vs. 持锁发送：临界区边界怎么划 (design): 采用「锁内快照 + 锁外发送」，既满足 map 访问互斥，又避免把 NIXL 慢调用放进临界区；快照还顺带消除了持活引用遍历的迭代风险。
• 淘汰路径加锁粒度：只让 pop 进临界区 (design): 以最小的临界区换取 _remote_agents 的不变式，锁注释「protects _handshake_futures and _remote_agents」从口头约定变为代码事实。
• 回归测试如何证明「同步」而不只是证明「行为」 (testing): 两条回归用例直接钉住锁语义，未加锁的实现会因 assert not done.wait(timeout=0.2) 失败，具备真正的回归价值；代价是引入固定超时的轻微时序依赖。
• fork PR 的自动 review 被跳过，仅人工 approve (other): 变更未经自动化深度审查即合入，技术论证主要依赖 PR body 与 commit message 的自述；对并发补丁而言，建议后续显式请求一次人工或机器人深审。
• CI 首次运行有 3 个 job 失败并触发重试 (testing): 重试后 PR 合入，未定位失败 job 是否为本次改动引起的真实回归；结合仓库近期多条 CI 修复 PR，更可能是环境或既有 flaky 用例。

风险与影响

• 风险：
  1. 残留 TOCTOU 窗口：快照在锁内完成，但 send_notif 在锁外执行。若淘汰在快照之后立即发生，writer 仍可能向刚被 remove_remote_agent 摘除的 agent 发通知（此前是「读到空 map 直接判失败」，现在变成「读到旧快照后再发送」）。发送失败会被 push_reg_notif_failed 记录，但若 NIXL 侧对已移除 agent 静默丢弃，则注册通知可能无人接收，需确认 NIXL 语义。
  2. 锁类型与潜在死锁：_cleanup_remote_engine 现在会获取 _handshake_lock，若基类中该锁是普通 Lock 而非 RLock（测试 stub 用的是 threading.RLock()），且存在「已持锁路径再调用 _cleanup_remote_engine」的调用链，就可能自锁死。新增测试通过直接 acquire()/release() 操纵锁，也说明该锁支持手动获取，但 RLock 与 Lock 的差异需以基类实现为准。
  3. 既有断言未加锁：assert engine_id in self._remote_agents 仍在临界区之外，并发重复淘汰同一引擎时 _remote_agents.pop(engine_id)（无默认值）仍可能抛 KeyError；这是改动前就存在的问题，未被本次修复覆盖。
  4. 测试时序敏感性：test_agent_read_is_serialized_with_handshake_lock 依赖 assert not done.wait(timeout=0.2)，在高负载 CI 上属于「依赖线程确实被阻塞」的判断，方向上是安全的（若未加锁会快速完成并使断言失败），但仍引入了固定超时带来的抖动因素。
  5. 回归面：改动只影响 NixlPushMode 的注册发送与引擎淘汰路径，未命中通用 NIXL 拉取路径，默认 TTL 为 0（_engine_ttl <= 0 时直接返回）时行为完全不变，因此对不启用淘汰的部署无感知。
     - 影响：对用户：使用 NixlPushMode 且 D 侧引擎频繁上下线（rollout、autoscaling）的部署，偶发的 No remote agents for engine ... 注册失败与随之而来的请求失败/重试会显著减少，属于稳定性与成功率改善，而非吞吐提升。对系统：_remote_agents 的并发不变式终于与锁注释一致，同一把 _handshake_lock 现在同时守护 handshake futures、agent map 的插入与删除，使 NIXL 连接器的并发模型更易推理，也为后续 X 系列（S1 淘汰、D churn）修复提供了可复用的锁模式。对团队：改动小、可读性高、回归测试可直接作为「如何证明同步语义」的模板；由于 review 仅 1 个 approve 且无评论，知识沉淀主要依赖 commit message 与测试 docstring。
     - 风险标记：并发核心路径变更, 残留 TOCTOU 发送窗口, 锁类型依赖 RLock 语义, 时序敏感并发测试, review 覆盖不足（无实质性评论）

关联脉络

• PR #55027 [Bugfix][KV Connector] MooncakeStore: exclude non-prefix-cacheable (QSA ring) groups; fix align-mode check: 同为 KV connector 子系统的并发/状态一致性 bugfix，同样改动 connector 的 scheduler/worker 侧并配套 tests/v1/kv_connector/unit 用例，可与本 PR 对照理解连接器侧的锁与状态清理约定。
• PR #56671 [CI][Test] Mock CPU backend block sizes in kv_connector unit conftest: 改的是 tests/v1/kv_connector/unit/conftest.py，正是本 PR 新增并发回归测试所依赖的单测夹具环境；该修复解阻了该目录下 81 个测试，是本 PR 测试能稳定跑通的前置条件。