执行摘要

• 一句话：修复 P2P 连接回收后的 EngineCore 崩溃
• 推荐动作：此 PR 值得精读，展示了系统化调试和分层修复的方法论。三个缺陷的独立识别、每个缺陷的单独验证（对照表中 solo fix 仍然崩溃）、以及测试先行的回归防护，都是高质量 Bug 修复的典范。建议团队将此 PR 作为关键系统代码审查的教学案例。

功能与动机

修复 #49809 描述的 EngineCore 崩溃问题：当请求重新连接到刚刚被 P2PSecondaryTierManager 回收的对等节点时，ZmqTransport.connect() 因为死连接仍注册在 _connections 中而触发 'AssertionError: ZmqConnection to :7777 already exists'。PR body 指出三个缺陷，任何单个修复都不足以解决。

实现拆解

1. 分离 _dead 与 _closed 标志（zmq.py） ZmqConnection 原有单一 _closed 同时被 mark_dead 和 close 使用，导致 close() 在 mark_dead 后提前返回，不释放 DEALER 和 monitor socket。现拆分为 _dead（对端死亡）和 _closed（资源已释放），mark_dead() 只翻转 _dead，close() 始终执行资源释放。alive 属性改为 not (self._dead or self._closed)，send() 守卫改为 not self.alive。此修复阻止 socket 泄漏，并确保 inproc 端点地址不冲突（因未释放的 monitor 仍持有）。

2. connect() 退役死连接（zmq.py） 原 connect() 直接用 assert peer_id not in self._connections 拒绝所有重复连接。现先检查是否有已注册的死连接，若有则先 pop 并 close() 之，再创建新连接；若已有活跃连接仍断言。此修复允许在死连接尚未被 poll() 清理的时间窗口内重连。

3. 单调递增 _monitor_seq（zmq.py） 将 inproc monitor 端点从基于 peer_id 改为基于 self._monitor_seq 单调计数器（在 _open_connection() 中自增并拼入端点地址），避免 libzmq 异步释放 inproc 端点导致的 EADDRINUSE 竞争。

4. poll() 清理顺序调整（zmq.py） 将死连接清理从 poll() 末尾移到开头，提取为 _sweep_dead_connections() 方法，并在 _check_monitors() 之后也调用一次，确保消息不会被注册到死连接中，也防止连接在 poll 周期内长期残留。

5. 文档契约明确（base.py） 在 ControlConnection.mark_dead 文档中添加“资源未在此释放，必须调用 close()”；在 ControlTransport.connect 文档中添加“允许用新连接替换注册的死连接”。

6. 测试扩展（test_zmq_transport.py） 新增 TestZmqReconnect 测试类，包含 4 个实数 socket 的测试用例，分别验证：close 在 mark_dead 后释放 socket、connect 替换死连接、同一对端 10 次重连不失败、poll 中死连接清理后消息不丢。同时在 test_dead_connection_removed_on_poll 中增加 socket closed 断言。所有测试纯 CPU 可运行。

关键文件：

• tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py（模块 测试；类别 test；类型 test-coverage；符号 TestZmqReconnect, test_close_after_mark_dead_releases_sockets, test_connect_retires_dead_connection, test_repeated_reconnect_to_same_peer）: 新增 108 行测试代码，包含完整重连场景覆盖，是验证三个缺陷修复的核心依据。每个测试在不含修复时均失败。
• vllm/v1/kv_offload/tiering/p2p/control/zmq.py（模块 控制连接；类别 source；类型 core-logic；符号 _sweep_dead_connections）: 核心修复文件，包含 _dead/_closed 分离、connect 退役死连接、_monitor_seq、_sweep_dead_connections 等所有逻辑变更。
• vllm/v1/kv_offload/tiering/p2p/control/base.py（模块 抽象接口；类别 source；类型 core-logic）: 文档契约更新，明确 mark_dead 不释放资源、connect 允许替换死连接。

关键符号：_sweep_dead_connections, ZmqConnection.mark_dead, ZmqConnection.close, ZmqTransport.connect, ZmqTransport.poll

关键源码片段

vllm/v1/kv_offload/tiering/p2p/control/zmq.py

核心修复文件，包含 _dead/_closed 分离、connect 退役死连接、_monitor_seq、_sweep_dead_connections 等所有逻辑变更。

# 摘自 ZmqConnection 的生命周期管理，修复 defect 2：
# mark_dead 和 close 使用独立标志，避免 socket 泄漏。
​
class ZmqConnection(ControlConnection):
    def __init__(self, peer_id: str, sockets: _Sockets) -> None:
        super().__init__(peer_id)
        self._sockets = sockets
        # _dead: 对端已死（逻辑断开）
        # _closed: 本端资源（socket）已释放
        self._dead = False
        self._closed = False
        self._inbox: list[dict] = []
​
    @property
    def alive(self) -> bool:
        # 返回 True 仅当既未标记为死连接也未调用 close
        return not (self._dead or self._closed)
​
    def mark_dead(self) -> None:
        # 标记连接为断开。只翻转 liveness，不释放资源。
        self._dead = True
​
    def close(self) -> None:
        # 释放所有资源（DEALER + monitor socket）。幂等。
        if self._closed:
            return
        self._closed = True
        self._dead = True
        self._sockets.monitor.close()
        self._sockets.dealer.close()

评论区精华

Review 中 @liranschour 建议将 connect() 中活跃重复连接的 assert 改为 RuntimeError（评论），但 PR 未采纳该提议，保留 assert 用于检测真正的逻辑 Bug。最终 reviewers 批准合并。

• connect 中活跃重复连接检查应该使用 RuntimeError 而非 assert (design): PR 作者未采纳，保留 assert。以 assert 作为逻辑 Bug 的快速失败机制，适用于开发环境；生产环境不期望发生此类情况。

风险与影响

• 风险：
  • 核心路径变更：_sweep_dead_connections 被提取并在 poll 中调用两次，可能对频繁 poll 的场景有微小性能影响，但死连接检测是 O(n)，n 为连接数，通常很小。
  • 生命周期重构：分离 _dead/_closed 改变了连接状态机，可能影响其他依赖于单一 _closed 标志的代码（无其他使用路径）。
  • inproc 端点唯一性依赖计数器 _monitor_seq，单调递增且不重置，在 long-running 服务中不会溢出（Python int 无上限）。
  • 测试覆盖：新增测试覆盖了核心重连场景，但未覆盖 monitor-late 或 silent-death 场景（PR 文档已明确为已知剩余窗口）。
• 影响：
  • 影响范围：仅限于 KV Offload P2P 场景的控制连接生命周期管理，不影响其他传输层（如 IPC、TCP）或功能路径。
  • 用户影响：修复了生产环境中 EngineCore 崩溃，使用 OffloadingConnector + P2P secondary tier 的用户可直接受益。
  • 兼容性：接口保持向后兼容，ControlConnection 和 ControlTransport 抽象方法签名不变，仅文档补充。
  • 性能：额外开销可忽略。
  • 风险标记：核心路径变更, ZMQ 生命周期调整, inproc 端点竞争, 测试覆盖同步

关联脉络

• 暂无明显关联 PR