执行摘要

• 一句话：修复分块预填充期间滑动窗口注意力存储可达性计算错误。
• 推荐动作：该 PR 值得精读，因为它修复了 KV 卸载中的一个正确性 bug，涉及复杂的调度状态机逻辑。设计决策中如何区分预填充、完成和中止状态，以及如何通过可达性掩码确保块存储的精确性，具有学习价值。建议关注 _build_store_jobs 中状态判断的逻辑流。

功能与动机

PR body 明确指出：'During chunked prefill, the offloading scheduler used the current step's SWA chunk count as the store-reachability horizon. This treated each intermediate frontier as the prompt's final partial segment and submitted chunks that the load path could never request.' 这导致存储的 SWA 块数量不匹配（例如 152 vs 预期 150），并在第一个不匹配的块 248 处出现。

实现拆解

1. 调度器核心逻辑修正：在 vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py 中新增 _reachable_store_block_mask 和 _final_swa_alignment_blocks 辅助方法，封装可达性掩码构建和最终 SWA 对齐块数计算。修改 _build_store_jobs 方法，引入 prompt_offloadable_tokens（提示最终可卸载令牌数）和 store_horizon_chunks 逻辑：预填充阶段使用提示水平，请求完成时使用当前水平，中止时重新考虑 SWA 尾块并避免重复提交。
2. 可达性掩码扩展：在 vllm/v1/core/single_type_kv_cache_manager.py 中，为 SlidingWindowManager.reachable_block_mask 类方法添加 final_segment_end_block 参数，使其能处理非 EAGLE SWA 组的最终部分段尾块可达性计算，确保在预填充未完成时正确标记可达块。
3. 测试覆盖增强：在 tests/v1/kv_connector/unit/offloading_connector/test_scheduler.py 中新增 test_chunked_prefill_uses_final_swa_store_horizon（覆盖完成、同步中止、异步中止三种生命周期）、test_active_decode_does_not_advance_swa_final_horizon（验证活跃解码不推进最终水平）和 test_dcp_abort_reconsiders_final_swa_tail（测试 DCP 中止场景）。在 tests/v1/core/test_prefix_caching.py 中新增 test_swa_reachable_block_mask_final_partial_segment 验证可达性掩码对最终部分段的正确性。

关键文件：

• vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py（模块 KV卸载调度器；类别 source；类型 core-logic；符号 _reachable_store_block_mask, _final_swa_alignment_blocks）: 核心逻辑修改，修复 SWA 存储可达性计算
• vllm/v1/core/single_type_kv_cache_manager.py（模块 KV缓存管理器；类别 source；类型 core-logic）: 扩展 reachable_block_mask 以支持最终段
• tests/v1/kv_connector/unit/offloading_connector/test_scheduler.py（模块 调度器测试；类别 test；类型 test-coverage；符号 test_chunked_prefill_uses_final_swa_store_horizon, abort_after_prefill_step, test_active_decode_does_not_advance_swa_final_horizon, test_dcp_abort_reconsiders_final_swa_tail）: 新增测试用例覆盖修复场景
• tests/v1/core/test_prefix_caching.py（模块 前缀缓存测试；类别 test；类型 test-coverage；符号 test_swa_reachable_block_mask_final_partial_segment, reachable_blocks）: 测试可达性掩码的正确性

关键符号：_reachable_store_block_mask, _final_swa_alignment_blocks, reachable_block_mask

关键源码片段

vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py

核心逻辑修改，修复 SWA 存储可达性计算

def _reachable_store_block_mask(
    self,
    group_config: GroupOffloadConfig,
    start_chunk_idx: int,
    end_chunk_idx: int,
    final_segment_end_chunk_idx: int | None,
    reachable_boundaries: tuple[int, ...],
) -> list[bool] | None:
    """Build the block mask for a range of candidate offload chunks."""
    blocks_per_chunk = self.config.blocks_per_chunk
    # 转换块索引并委托给 manager_cls 的 reachable_block_mask
    return group_config.manager_cls.reachable_block_mask(
        start_block=start_chunk_idx * blocks_per_chunk,
        end_block=end_chunk_idx * blocks_per_chunk,
        alignment_tokens=self.config.alignment_tokens,
        kv_cache_spec=group_config.kv_cache_spec,
        use_eagle=group_config.is_eagle_group,
        retention_interval=self.config.retention_interval,
        reachable_boundaries=reachable_boundaries,
        dcp_world_size=self.config.dcp_world_size,
        # 传递最终段结束块索引，用于非 EAGLE SWA 组的尾块可达性计算
        final_segment_end_block=(
            final_segment_end_chunk_idx * blocks_per_chunk
            if final_segment_end_chunk_idx is not None
            else None
        ),
    )
​
def _final_swa_alignment_blocks(
    self, group_config: GroupOffloadConfig
) -> int | None:
    """Return representable final-tail alignment for a non-EAGLE SWA group."""
    alignment_tokens = self.config.alignment_tokens
    if (
        alignment_tokens is None
        or group_config.is_eagle_group
        or not isinstance(group_config.kv_cache_spec, SlidingWindowSpec)
        or alignment_tokens % group_config.tokens_per_block != 0
    ):
        return None
    return alignment_tokens // group_config.tokens_per_block

修改后的 _build_store_jobs 核心逻辑：

def _build_store_jobs(
    self,
    scheduler_output: SchedulerOutput,
) -> dict[int, TransferJob]:
    blocks_per_chunk = self.config.blocks_per_chunk
    store_jobs: dict[int, TransferJob] = {}
    for req_id in chain(
        scheduler_output.num_scheduled_tokens,
        scheduler_output.finished_req_ids or (),
    ):
        req_status = self._req_status.get(req_id)
        if req_status is None:
            continue
        req = req_status.req
​
        # 根据请求状态计算块数
        if req.status is RequestStatus.FINISHED_ABORTED:
            num_tokens_after_batch = req.num_computed_tokens
        elif req.is_finished():
            num_tokens_after_batch = req.num_tokens
        else:
            num_scheduled_tokens = scheduler_output.num_scheduled_tokens[req_id]
            num_tokens_after_batch = req.num_computed_tokens + num_scheduled_tokens
​
        num_offloadable_tokens = self._calc_num_offloadable_tokens(
            req_status, num_tokens_after_batch
        )
        # 新增：计算提示的最终可卸载令牌数
        prompt_offloadable_tokens = self._calc_num_offloadable_tokens(
            req_status, req.num_prompt_tokens
        )
        # ... 后续循环中为每个 group_config 计算 store_horizon_chunks 和 reconsider_final_swa_tail ...

vllm/v1/core/single_type_kv_cache_manager.py

扩展 reachable_block_mask 以支持最终段

@classmethod
def reachable_block_mask(
    cls,
    start_block: int,
    end_block: int,
    alignment_tokens: int | None,
    kv_cache_spec: KVCacheSpec,
    use_eagle: bool,
    retention_interval: int | None = None,
    reachable_boundaries: Sequence[int] = (),
    dcp_world_size: int = 1,
    final_segment_end_block: int | None = None, # 新增参数
) -> list[bool] | None:
    assert isinstance(kv_cache_spec, SlidingWindowSpec)
    if alignment_tokens is None:
        return None
    block_size = kv_cache_spec.block_size * dcp_world_size
    if alignment_tokens % block_size != 0:
        return None
​
    need = cls._contiguous_blocks_for_hit(
        window_size=kv_cache_spec.sliding_window,
        block_size=block_size,
        use_eagle=use_eagle,
    )
    shift = 1 if use_eagle else 0
    mask = [False] * (end_block - start_block)
​
    # (1) Segment-boundary tails.
    segment_tokens = (
        alignment_tokens
        if retention_interval is None
        else (None if retention_interval == 0 else retention_interval)
    )
    if segment_tokens is not None:
        per_segment = segment_tokens // block_size
        if need >= per_segment:
            return None
        for i in range(start_block, end_block):
            if i >= shift and (i - shift) % per_segment >= per_segment - need:
                mask[i - start_block] = True
        # 新增：处理最终段尾块可达性
        if final_segment_end_block is not None and not use_eagle:
            final_segment_size = final_segment_end_block % per_segment
            if final_segment_size:
                final_segment_start = final_segment_end_block - final_segment_size
                final_tail_start = max(
                    final_segment_start, final_segment_end_block - need
                )
                final_tail_end = min(end_block, final_segment_end_block)
                for i in range(max(start_block, final_tail_start), final_tail_end):
                    mask[i - start_block] = True
​
    # (2) Reachable-boundary tails.
    if retention_interval is not None:
        for boundary_tokens in reachable_boundaries:
            aligned = boundary_tokens // alignment_tokens * alignment_tokens
            end = aligned // block_size + shift
            for j in range(max(start_block, end - need), min(end_block, end)):
                mask[j - start_block] = True
​
    return mask

评论区精华

orozery：'Thanks @Whamp ! I'm planning to merge #51886 first, then let's see how we can work on top of it.'

Whamp：'Thanks @orozery. #51886 has landed, so I rebased this PR onto current main and adapted the fix to the new reachable_block_mask path.'

讨论围绕合并顺序展开，无技术争议。PR 作者在 #51886 合并后 rebase 并适配了新的 reachable_block_mask 路径，解决了依赖关系。

• 合并顺序与依赖 (design): PR 作者在 #51886 合并后 rebase 到 main，并适配了新的 reachable_block_mask 路径。

风险与影响

• 风险：变更涉及 KV 卸载调度器的核心路径 _build_store_jobs，影响所有使用滑动窗口注意力和 KV 卸载的模型（如某些多模态模型）。主要风险包括：

  • 正确性回归：新引入的 store_horizon_chunks 和 reconsider_final_swa_tail 逻辑若存在边界错误，可能导致存储块集合不正确，影响 KV 卸载的完整性和加载效率。
  • 性能影响：额外计算 prompt_offloadable_tokens 和可达性掩码可能引入微小开销，但对整体推理性能影响有限。
  • 兼容性：reachable_block_mask 新增参数可能影响现有调用点，但代码显示默认值为 None，保持向后兼容。
  • 测试覆盖：新增测试覆盖关键场景，但未进行真实 GPU 服务评估，生产环境需观察。
  • 影响：用户影响：透明修复，用户无需调整代码，但 KV 卸载的可靠性提升，避免因存储不可达块导致的内存浪费或加载失败。
    系统影响：改进 KV 卸载调度器的正确性，确保分块预填充期间 SWA 块存储与请求状态严格同步，降低系统级错误概率。
    团队影响：提供了更健壮的测试框架，为后续 KV 卸载优化奠定基础，增强团队对复杂调度逻辑的信心。

• 风险标记：核心路径变更, 测试覆盖增加, 模型兼容性风险

关联脉络

• PR #51886 Fix shared block pool DCP scaling (推断标题): PR 作者明确提到 rebase 到 #51886 之后，并适配了其新的 reachable_block_mask 路径，表明两者在 KV 卸载可达性逻辑上紧密相关。