# PR #47021 完整报告

- 仓库：`vllm-project/vllm`
- 标题：[NIXL] Avoid reading expired blocks in bidirectional turn-2 read
- 合并时间：2026-07-14 21:03
- 原文链接：http://prhub.com.cn/vllm-project/vllm/pull/47021

---

# 执行摘要

- 一句话：阻止双向模式下 P 读取 D 端过期 KV blocks
- 推荐动作：值得精读，尤其是时钟同步策略、安全裕度设计以及如何在不引入过多配置项的情况下保证正确性。对于分布式 KV 传输贡献者，本 PR 展示了以最小侵入性解决数据一致性问题的方法。

# 功能与动机

来自 PR 描述：当前多轮双向模式下，D 为 KV blocks 设置固定 TTL，但 heartbeat 不会延长该 TTL，因此 blocks 过期后可被其他请求复用。后续 turn 无法识别该 blocks 是否有效，导致读取到乱码。本 PR 通过让 D 提交 blocks 过期时间，P 在读取前校验，解决该问题。

# 实现拆解

变更分五个核心步骤：
1. **版本 Bump 与元数据扩展**：在 `metadata.py` 中将 `NIXL_CONNECTOR_VERSION` 从 4 提升到 5，并在 `RemoteMeta` 中添加 `blocks_expiry_time: float | None` 字段，在 `add_new_req_to_recv` 中从 `kv_transfer_params` 读取 `remote_blocks_expiry_time`。
2. **Handshake 时钟同步**：修改 `base_worker.py` 中的 `_nixl_handshake`，在发送握手消息前记录 `time.perf_counter()`，收到回复后解析第二个帧（D 端的 `perf_counter`），计算 RTT 并取最小 RTT 样本对应的时钟偏移 `best_offset`。同时，`pull_scheduler.py` 中的 `request_finished` 在 D 节点上导出本地 `blocks_expiry_time` 为 `remote_blocks_expiry_time`。
3. **时钟偏移写入**：将计算出的偏移存入 `base_worker.py` 新增的 `_engine_clock_offset` 字典，该字典在 `done_callback` 中（以线程安全方式）写入，而非直接在 handshake 后台线程中写入。
4. **过期校验逻辑**：在 `pull_worker.py` 中新增 `_is_turn2_read_expired` 方法，将 D 端 `blocks_expiry_time` 减去本地估计的时钟偏移，并与当前 `perf_counter` 加上安全裕度（5 秒）比较；如果过期或即将过期则返回 `True`。在 `_read_blocks_for_req` 开头检查该条件，若为真则记录 `xfer_stats` 并调用 `_handle_failed_transfer` 拒绝本次读取。
5. **双向模式门控与测试**：整个过期检查逻辑仅在 `_bidirectional_kv_xfer_enabled` 为 `True`（通过配置 `bidirectional_kv_xfer` 启用）时执行。同时，`test_bidirectional_kv_transfer.py` 新增 4 个测试用例覆盖偏移、过期、安全裕度、缺失 deadline 等场景，并调整已有辅助函数以适配新版本 handshake 签名（返回双值）。
配套变更包括 `base_scheduler.py` 中 handshake listener 回显 `perf_counter`，`push_worker.py` 适配签名，`test_nixl_connector.py` 和 `test_nixl_connector_hma.py` 适配参数变化，以及 `docs/design/nixl_kv_cache_lease.md` 更新文档说明。

关键文件：
- `tests/v1/kv_connector/unit/test_bidirectional_kv_transfer.py`（模块 双向传输；类别 test；类型 test-coverage；符号 test_turn2_deadline_gate, test_turn2_full_prefix_hit_with_expired_deadline_skips_gate, test_d_node_request_finished_exports_blocks_expiry_time, test_handshake_listener_appends_perf_counter_frame）: 新增 4 个测试用例覆盖 turn-2 deadline 校验的各类场景（过期、偏移导致过期、安全裕度内、远未过期、缺失 deadline），确保 P 端正确拒绝或放行读取。同时调整辅助函数以适配 handshake 返回 signature 变化。
- `vllm/distributed/kv_transfer/kv_connector/v1/nixl/base_worker.py`（模块 分布式；类别 source；类型 core-logic；符号 done_callback）: 核心改动：启用双向模式配置标志；新增 _engine_clock_offset 字典；修改 _nixl_handshake 以测量 RTT 和时钟偏移；在 done_callback 中安全写入偏移。
- `vllm/distributed/kv_transfer/kv_connector/v1/nixl/pull_worker.py`（模块 分布式；类别 source；类型 core-logic；符号 _is_turn2_read_expired）: 新增过期检查核心函数 _is_turn2_read_expired，并在 _read_blocks_for_req 开头调用；只有双向模式启用时才执行。
- `vllm/distributed/kv_transfer/kv_connector/v1/nixl/pull_scheduler.py`（模块 调度器；类别 source；类型 core-logic）: 在 request_finished 中，当 is_d_node 为 True 时，导出 blocks_expiry_time 作为 remote_blocks_expiry_time 传递给 P。
- `vllm/distributed/kv_transfer/kv_connector/v1/nixl/metadata.py`（模块 元数据；类别 source；类型 data-contract）: 版本 bump 4->5，添加 RemoteMeta.blocks_expiry_time 字段，并在 add_new_req_to_recv 中解析。

关键符号：_is_turn2_read_expired, _nixl_handshake, _nixl_handshake_listener, request_finished, add_new_req_to_recv, done_callback

## 关键源码片段

### `tests/v1/kv_connector/unit/test_bidirectional_kv_transfer.py`

新增 4 个测试用例覆盖 turn-2 deadline 校验的各类场景（过期、偏移导致过期、安全裕度内、远未过期、缺失 deadline），确保 P 端正确拒绝或放行读取。同时调整辅助函数以适配 handshake 返回 signature 变化。

```python
@pytest.mark.parametrize(
    ("offset", "expiry_delta", "expect_declined"),
    [
        # 已过期：D 端 deadline 比当前时间早 10s → 应拒绝
        pytest.param(0.0, -10.0, True, id="expired"),
        # 时钟偏移致过期：P 认为 D 快 100s，D 的 deadline 还有 50s → 相对 P 已过期
        pytest.param(100.0, 50.0, True, id="offset_makes_expired"),
        # 在 5s 安全裕度内 → 视为可能过期，拒绝
        pytest.param(0.0, 2.0, True, id="near_expiry_within_margin"),
        # 未来很远 → 有效
        pytest.param(0.0, 1000.0, False, id="valid_far_deadline"),
        # 无 deadline（旧版本节点） → 跳过，正常读取
        pytest.param(0.0, None, False, id="missing_expiry_field"),
    ],
)
@patch("vllm.distributed.kv_transfer.kv_connector.v1.nixl.base_worker.NixlWrapper",
       FakeNixlWrapper)
def test_turn2_deadline_gate(dist_init, offset, expiry_delta, expect_declined):
    """校验 P 在 turn-2 读取时根据 deadline 拒绝或放行的行为。"""
    connector, worker = _make_connector_with_fake_worker()
    worker._engine_clock_offset[_REMOTE] = offset  # 注入 P 对 D 的时钟偏差
    # 构造 deadline：如果 expiry_delta 为 None 则缺失 deadline
    expiry_time = None if expiry_delta is None else time.perf_counter() + expiry_delta
    meta = NixlConnectorMetadata()
    params = {
        "do_remote_prefill": False,
        "do_remote_decode": True,
        "remote_block_ids": ([20, 21],),
        "remote_engine_id": FakeNixlConnectorWorker.REMOTE_ENGINE_ID,
        "remote_request_id": "decode-req",
        "remote_host": "localhost",
        "remote_port": 1234,
        "remote_tp_size": 1,
    }
    if expiry_time is not None:
        params["remote_blocks_expiry_time"] = expiry_time
    meta.add_new_req_to_recv(
        request_id="req",
        local_block_ids=([10, 11],),
        kv_transfer_params=params,
    )
    _do_load_kv(connector, meta)
    # 检查统计数据：应该拒绝（增加 num_kv_expired_reqs）或正常
    expected_expired_count = [1] if expect_declined else [0]
    assert worker.xfer_stats.data["num_kv_expired_reqs"] == expected_expired_count

```

### `vllm/distributed/kv_transfer/kv_connector/v1/nixl/pull_worker.py`

新增过期检查核心函数 _is_turn2_read_expired，并在 _read_blocks_for_req 开头调用；只有双向模式启用时才执行。

```python
# 安全裕度（秒），用于吸收时钟偏移误差和读取延迟
_KV_BLOCKS_EXPIRY_SAFETY_MARGIN = 5.0

class NixlPullConnectorWorker(NixlBaseConnectorWorker):
    # ... 其他代码

    def _is_turn2_read_expired(self, meta: ReqMeta) -> bool:
        """判断 D 端缓存的 blocks 对本次 turn-2 读取是否已（接近）过期。"""
        assert meta.remote is not None
        blocks_expiry_time = meta.remote.blocks_expiry_time
        # 若 deadline 不存在（如旧版本或 router 未转发），则正常读取
        if blocks_expiry_time is None or not meta.local_physical_block_ids:
            return False
        # 从当前记录的偏移中查找对应 D 引擎的时钟偏移
        clock_offset = self._engine_clock_offset[meta.remote.engine_id]
        # 将 D 域的 deadline 转换为本地域时间
        deadline = blocks_expiry_time - clock_offset
        # 当前时间加上裕度 >= deadline 视为过期
        return time.perf_counter() + _KV_BLOCKS_EXPIRY_SAFETY_MARGIN >= deadline

    def _read_blocks_for_req(self, req_id: str, meta: ReqMeta):
        assert meta.remote is not None and self.transfer_topo is not None
        engine_id = meta.remote.engine_id
        self._engine_last_active[engine_id] = time.perf_counter()
        # 双向模式延迟退出：若 blocks 已过期则拒绝读取
        if self._bidirectional_kv_xfer_enabled and self._is_turn2_read_expired(meta):
            logger.warning(
                "Declining expired remote read for %s from engine %s.",
                req_id,
                engine_id,
            )
            self.xfer_stats.record_kv_expired_req()
            self._handle_failed_transfer(req_id, None)
            return
        # ... 后续正常读取流程

```

# 评论区精华

- **安全裕度是否可配置**：NickLucche 认为应减少配置项，使用硬编码 5 秒即可。最终移除该配置项，改为模块级常量 `_KV_BLOCKS_EXPIRY_SAFETY_MARGIN`。
- **RTT 最小化策略**：NickLucche 要求明确为什么取最小 RTT 样本，解释为同一集群内 rank 间传输延迟近似，高 RTT 为噪声，使用最小值可使时钟偏移估计更准确。最终添加了注释说明。
- **时钟偏移写入位置**：NickLucche 指出不应在 handshake 线程中直接写入 `_engine_clock_offset`，应返回偏移值并在 `done_callback` 中写入，因为 `_engine_clock_offset` 在主线程读取，需避免竞争。最终调整。
- **多部分消息断言**：NickLucche 建议 assert `len(reply_parts) == 2`，因为版本 bump 已破坏向后兼容，可以直接要求两个帧。最终添加了断言。
- **逻辑提取与门控**：NickLucche 建议将过期检查提取为独立方法 `_is_turn2_read_expired`，并仅在 `_bidirectional_kv_xfer_enabled` 启用时调用，以减少对常规流程的认知负担。作者最终提取了该方法。

 - 安全裕度是否应暴露为配置 (design): 不暴露，使用模块级常量 `_KV_BLOCKS_EXPIRY_SAFETY_MARGIN = 5.0`。
 - RTT 最小化策略的设计理由 (design): 添加注释说明选择最小 RTT 的原因。
 - 时钟偏移写入位置应放在 done_callback 中 (correctness): 修改为在 `done_callback` 中写入偏移，确保主线程读取时一致。
 - 过期检查应提取为独立方法并双边门控 (testing): 提取为独立方法，门控在双向模式配置下。

# 风险与影响

- 风险：
 - **时钟同步误差**：RTT 最小化策略仍可能因网络波动导致偏移估计偏差，但 5 秒安全裕度提供了缓冲，且测试覆盖了多种偏移场景，风险较低。
 - **性能影响**：handshake 增加一次 `perf_counter` 回复，但仅在连接时发生，对运行时无影响；过期检查在每次 read 调用时增加一次浮点比较，可忽略。
 - **不兼容性**：版本号 bump 4→5 意味着旧版 NIXL 节点无法通信，但由于组件通常一起升级，影响可控。
 - **潜在竞态**：`done_callback` 与主线程可能并发访问 `_engine_clock_offset`，但主线程读取时假设 handshake 已完成，当前使用 `RLock` 保护 handshake futures，读取时未加显式锁，需要确认无竞态。
- 影响：
 - **用户**：多轮对话质量显著提升，不再因读取过期 block 出现乱码；需启用 `bidirectional_kv_xfer` 新配置。
 - **系统**：handshake 阶段增加一次额外消息，但频率极低；验证逻辑新增在 `_read_blocks_for_req` 中，对吞吐影响可忽略。
 - **团队**：简化了配置接口（未暴露 safety margin），使代码更易维护；单元测试覆盖了关键路径，降低后续回归风险。
 - 风险标记：时钟同步误差 , 安全裕度硬编码 , 版本兼容性破坏 , 潜在竞态

# 关联脉络

- 暂无明显关联 PR