Bug 原理

根因机制

reset_cache() 用于清空 offloading scheduler 的所有 cache 状态（通常在引擎重启、重配置或需要驱逐所有 stored blocks 时调用）。它做了三件事：

  1. _jobs.clear() — 清空 scheduler 侧所有 in-flight job 的状态记录
  2. _stale_job_threshold = _job_counter — 标记一个阈值，之后到达的 worker 完成响应如果 job_id < 阈值，则视为”过时响应”并丢弃
  3. 但遗漏了 status.transfer_jobs.clear() — 每个请求的 RequestOffloadState 中的 transfer_jobs（set[int]）没有被清空

后果链（为什么请求会永久卡死）

reset_cache() 调用
  │
  ├─ _jobs.clear()          → scheduler 侧 job 状态清空
  ├─ _stale_job_threshold   → 旧 job_id 被标记为过时
  └─ status.transfer_jobs   → ❌ 未清空，残留旧 job_id
         │
         ▼
  后续调度步骤调用 get_num_new_matched_tokens()
         │
         ├─ if req_status.transfer_jobs:  → 永远为 True（残留 job_id 永远在 set 里）
         ├─ return None, False            → 请求被 defer
         └─ 请求移入 step_skipped_waiting → 每步都被跳过
                │
                ▼
         请求永久卡死：
           - 永远不会被调度
           - 永远不会完成
           - request_finished() 永远不会被调用
           - _req_status 永远不释放（内存泄漏）
           - _build_load_jobs 中 assert not transfer_jobs → AssertionError

为什么这些残留 job_id 不会自然消失

正常情况下，transfer_jobs 中的 job_id 通过 update_connector_output() 路径移除：worker 完成传输后报告，scheduler 从 transfer_jobs 中 discard(job_id)。

但 reset_cache() 之后：

修复

一行代码：status.transfer_jobs.clear()，在 reset_cache() 中遍历存活的请求时清空其 transfer_jobs。

OffloadingConnectorScheduler 实现机制

角色定位

OffloadingConnectorScheduler（OCS）是 vLLM scheduler 侧的 offloading 管理器，位于 vllm/distributed/kv_transfer/kv_connector/v1/offloading/scheduler.py。它负责：

核心数据结构

OffloadingConnectorScheduler
├── _req_status: dict[ReqId, RequestOffloadState]  # 每个请求的 offloading 状态
│   └── RequestOffloadState
│       ├── transfer_jobs: set[int]       # 请求当前 in-flight 的 job ID 集合
│       ├── group_states: tuple[...]       # 每个 KV cache group 的状态（hash keys、block IDs 等）
│       ├── num_locally_computed_tokens: int  # GPU 本地计算的 token 数
│       └── max_offload_tokens: int|None  # 该请求最多 offload 的 token 上限
│
├── _jobs: dict[int, TransferJobStatus]  # 所有活跃 job 的状态
│   └── TransferJobStatus
│       ├── req_id: ReqId               # 属于哪个请求
│       ├── pending_count: int          # 还有几个 worker 未完成（0 = job 完成）
│       ├── keys: set[OffloadKey]       # 这个 job 覆盖的 offload key
│       └── is_store: bool              # 是 store 还是 load job
│
├── _job_counter: int                   # job ID 生成器（单调递增）
├── _stale_job_threshold: int           # 区分"旧 job"和"新 job"的阈值
├── _blocks_being_loaded: set[OffloadKey]|None  # 正在加载的 blocks（避免重复加载）
└── manager: OffloadingManager          # 实际执行 lookup/load/store 的后端抽象层

关键流程

1. get_num_new_matched_tokens() — 决定可加载的 token 数

输入：request, num_computed_tokens（GPU 已计算的 token 数）
输出：(num_new_tokens, is_async)

步骤：
  1. 检查 transfer_jobs 是否为空
     - 非空 → return None, False（请求被 defer，等待传输完成）
  2. 调用 _lookup() 向后端查询哪些 blocks 命中
  3. 返回可加载的 token 数

2. _lookup() — 向后端查询命中情况

3. update_connector_output() — worker 完成报告

输入：worker 报告的 KVConnectorOutput
处理：
  1. 遍历完成的 job_id
  2. 从 transfer_jobs 中移除该 job_id
  3. 调用 manager.complete_load/store() 通知后端
  4. 如果是最后一个 worker（pending_count == 0）→ job 真正完成

4. reset_cache() — 清空所有 cache 状态

调用场景：引擎重启、重配置、驱逐所有 stored blocks

步骤：
  1. 断言没有正在进行的调度步骤
  2. 标记所有现有 job 为"需要 flush"
  3. 删除已完成的请求（_req_status 中 is_finished 的）
  4. 调用 manager.reset_cache() 清空后端
  5. 重置存活的请求的 store 进度（next_stored_block_idx = 0）
  6. ✅ 清空 transfer_jobs（PR #48102 的修复）
  7. 更新 _stale_job_threshold，清空 _jobs 和 _block_id_to_pending_jobs

请求状态机

  [新请求]
      │ on_new_request()
      ▼
  [无传输] ─────────────────────────────┐
      │                                 │
      │ get_num_new_matched_tokens()     │ transfer_jobs 清空后
      │ 发现可加载 token                 │   回到此状态
      ▼                                 │
  [加载传输中]                           │
      │ transfer_jobs = {load_job_id}   │
      │                                 │
      │ update_connector_output()       │
      │ 加载完成                        │
      ▼                                 │
  [无传输] ─────────────────────────────┘
      │
      │ 需要 store（decode 产生新 blocks）
      ▼
  [存储传输中]
      │ transfer_jobs = {store_job_id_1, store_job_id_2, ...}
      │
      │ update_connector_output()
      │ 所有 store 完成
      ▼
  [无传输]

为什么这个 bug 是”可达的”