[KVCache][Feature] Implement Cache Manager V1 Storage Transfer and Prefetch (2/n)

[kevincheng2]

Motivation

本 PR 是 Cache Manager V1 系列的第二部分，基于 #7097 之上构建。

引入了 Cache Manager V1 的 storage transfer 子系统，主要能力包括：

• KV cache 从 GPU/CPU 卸载到外部存储（Mooncake、RDMA、IPC）
• 异步从外部存储预取 KV cache 回设备
• D2H → Storage 链式传输，实现多级缓存无缝驱逐
• 基于 ZMQ 的 Scheduler 与 Worker 之间的预取 pipeline

Modifications

核心模块

• fastdeploy/cache_manager/v1/storage/

  • staging_manager.py：新增 StagingManager，管理异步存储传输队列及预取任务
  • base.py：更新 StorageConnector 基类，增加 prefetch/backup 接口
  • mooncake/connector.py：Mooncake 存储连接器，支持 prefetch/backup

• fastdeploy/cache_manager/v1/transfer_manager.py：重构以支持多级传输（GPU↔CPU↔Storage），将 TransferManager 与 StagingManager 分离

• fastdeploy/cache_manager/v1/cache_controller.py：新增存储预取/备份编排逻辑及 D2H→Storage 链式传输

• fastdeploy/cache_manager/v1/cache_manager.py：集成存储传输 API

• fastdeploy/cache_manager/v1/block_pool.py：支持存储层级 block

Engine & Worker 集成

• fastdeploy/engine/sched/resource_manager_v1.py：在资源管理器中集成存储预取触发逻辑
• fastdeploy/worker/gpu_worker.py / worker_process.py：新增 ZMQ 预取 pipeline 处理
• fastdeploy/inter_communicator/zmq_client.py：扩展 ZMQ client 支持预取消息
• fastdeploy/engine/common_engine.py / request.py：存储传输元数据相关更新

测试

• tests/cache_manager/v1/test_staging_manager.py：StagingManager 单元测试
• tests/cache_manager/v1/test_transfer_manager.py：更新 transfer manager 测试
• tests/cache_manager/v1/test_cache_manager.py：更新 cache manager 测试
• tests/cache_manager/v1/test_cache_utils.py：更新 cache utils 测试

Usage or Command

启动包含 storage transfer 功能的服务示例：

# 编译算子（仅修改了 Python 代码，通常无需重新编译）
# bash build.sh 1 python false [90] 1

Checklist

• Add at least a tag in the PR title.
• Format your code, run pre-commit before commit.
• Add unit tests.
• Provide accuracy results.（不涉及模型前向代码修改，无需精度测试）
• If the current PR is submitting to the release branch, make sure the PR has been submitted to the develop branch.

[paddle-bot]

Thanks for your contribution!

[PaddlePaddle-bot]

🤖 Paddle-CI-Agent | ci_status_monitor | 2026-05-13 18:19:38

CI报告基于以下代码生成（30分钟更新一次）:

• PR commit: b4f54a9
• Merge base: ecce6b5 (branch: develop)
• 查看完整 Diff
• CI 详情

---

1 任务总览

⚠️ 1 个 Required 失败，2 个 Required 运行中，Approval 审批未通过，需处理后方可合并。

总执行（rerun次数）	总任务	✅ 通过	❌ 失败	⏳ 运行中	⏸️ 等待中	跳过
38(0)	38	31	4	2	1	0

2 任务状态汇总

2.1 Required任务 : 7/10 通过

必选任务阻塞合并，失败需优先处理。

状态	任务	耗时	根因	修复建议	日志	重跑
❌	Approval	9s	PR问题：PR修改日志行为，需指定RD成员审批	请联系 xyxinyang/zyyzghb 在PR上 Approve	Job	-
⏳	xpu_4cards_case_test / run_xpu_4cards_cases	-	运行中	-	Job	-
⏳	Run FastDeploy Unit Tests and Coverage / run_tests_with_coverage	-	运行中	-	Job	-
✅	其余 7 个必选任务通过	-	-	-	-	-

2.2 可选任务 — 24/28 通过

可选任务不阻塞合并，失败仅供参考。

状态	任务	耗时	日志	重跑
❌	Run iluvatar Tests / run_iluvatar_cases	10m34s	Job	-
❌	Check PR Template	17s	Job	-
❌	Trigger Jenkins for PR	25m47s	Job	-
⏸️	CI_HPU	-	-	-
✅	其余 24 个可选任务通过	-	-	-

---

3 失败详情（仅 required）

Approval — 代码规范（置信度: 高）

Approval

• 状态: ❌ 失败
• 错误类型: 代码规范
• 置信度: 高
• 根因摘要: PR新增大量logging调用，需FastDeploy指定RD成员审批
• 分析器: 通用分析(fallback)

根因详情:
PR (#7529) 在 diff 中新增了大量 logger.info / logger.debug / logger.error 调用，触发了 FastDeploy 的 check_approval.sh 审批检查。该检查要求修改日志行为的 PR 必须获得指定 FastDeploy RD 成员（xyxinyang(zhouchong) 或 zyyzghb(zhangyongyue)）的 GitHub Approve，退出码 6 表示当前尚未获得所需审批。

关键日志:

Detected log modification in diff
0. You must have one FastDeploy RD (xyxinyang(zhouchong), zyyzghb(zhangyongyue)) approval
   for modifying logging behavior (.info/.debug/.error/log_request).
There are 1 approved errors.
Process completed with exit code 6.

修复建议:

1. 请 PR 作者联系 xyxinyang(zhouchong) 或 zyyzghb(zhangyongyue) 在本 PR 上进行 GitHub Review Approve 操作
2. Approve 完成后重新触发 Approval workflow（Re-run failed jobs）

修复建议摘要: 请联系 xyxinyang/zyyzghb 在PR上 Approve

关联变更: PR 新增 StoragePrefetch、StorageBackup、MooncakeDistributedStore 等模块的大量 logger 调用

链接: 查看日志

[codecov-commenter]

Codecov Report

❌ Patch coverage is 31.21577% with 628 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (develop@ecce6b5). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
...loy/cache_manager/v1/storage/mooncake/connector.py	0.00%	296 Missing ⚠️
fastdeploy/cache_manager/v1/transfer_manager.py	22.36%	121 Missing and 4 partials ⚠️
fastdeploy/cache_manager/v1/cache_manager.py	36.20%	71 Missing and 3 partials ⚠️
fastdeploy/worker/worker_process.py	15.68%	40 Missing and 3 partials ⚠️
fastdeploy/engine/sched/resource_manager_v1.py	85.62%	0 Missing and 23 partials ⚠️
fastdeploy/cache_manager/v1/cache_controller.py	47.50%	18 Missing and 3 partials ⚠️
fastdeploy/inter_communicator/zmq_client.py	8.33%	11 Missing ⚠️
...oy/cache_manager/v1/storage/attnstore/connector.py	0.00%	9 Missing ⚠️
fastdeploy/cache_manager/v1/storage/base.py	36.36%	7 Missing ⚠️
fastdeploy/engine/request.py	57.14%	5 Missing and 1 partial ⚠️
... and 5 more

Additional details and impacted files

@@            Coverage Diff             @@
##             develop    #7529   +/-   ##
==========================================
  Coverage           ?   62.71%           
==========================================
  Files              ?      461           
  Lines              ?    64895           
  Branches           ?     9960           
==========================================
  Hits               ?    40697           
  Misses             ?    21383           
  Partials           ?     2815           

Flag	Coverage Δ
GPU	71.64% <31.21%> (?)
XPU	7.13% <6.90%> (?)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[PaddlePaddle-bot]

🤖 Paddle-CI-Agent | pr_review | 2026-05-13 18:19:39

📋 Review 摘要

PR 概述：实现 Cache Manager V1 的 storage transfer 子系统，包含 GPU↔CPU↔Storage 多级传输、D2H→Storage 链式驱逐、以及基于 ZMQ 的预取 pipeline。
变更范围：cache_manager/v1/、engine/sched/、worker/、inter_communicator/
影响面 Tag：[KVCache] [Engine] [Scheduler]

---

📝 PR 规范检查

标题使用了两个 Tag [KVCache][Feature]，但规范要求仅一个官方 Tag；PR body 缺少 ## Accuracy Tests 段落。

标题建议（可直接复制）：

• [KVCache] Implement Cache Manager V1 Storage Transfer and Prefetch (2/n)

PR 描述建议（可直接复制，需补充 ## Accuracy Tests 段落）：

## Motivation
本 PR 是 Cache Manager V1 系列的第二部分，基于 #7097 之上构建。

引入了 Cache Manager V1 的 **storage transfer** 子系统，主要能力包括：
- KV cache 从 GPU/CPU 卸载到外部存储（Mooncake、RDMA、IPC）
- 异步从外部存储预取 KV cache 回设备
- D2H → Storage 链式传输，实现多级缓存无缝驱逐
- 基于 ZMQ 的 Scheduler 与 Worker 之间的预取 pipeline

## Modifications
- `staging_manager.py`：新增 StagingManager，管理异步存储传输队列及预取任务
- `storage/base.py`：更新 StorageConnector 基类，增加 prefetch/backup 接口
- `storage/mooncake/connector.py`：Mooncake 存储连接器，支持 prefetch/backup
- `transfer_manager.py`：重构以支持多级传输（GPU↔CPU↔Storage）
- `cache_controller.py`：新增存储预取/备份编排逻辑及 D2H→Storage 链式传输
- `cache_manager.py`：集成存储传输 API，实现 prefetch_storage / abort_prefetch_blocks 等
- `engine/sched/resource_manager_v1.py`：集成存储预取触发逻辑
- `worker/gpu_worker.py`、`worker_process.py`：新增 ZMQ 预取 pipeline 处理
- `inter_communicator/zmq_client.py`：扩展 ZMQ client 支持预取消息

## Usage or Command
```bash
python -m fastdeploy.entrypoints.openai.api_server \
  --model /path/to/model \
  --cache-manager-version v1 \
  --host-cache-size-gb 40 \
  --storage-backend mooncake \
  --tensor-parallel-size 8
```

## Accuracy Tests
N/A（本 PR 不涉及模型前向代码修改，无需精度测试）

## Checklist

- [x] Add at least a tag in the PR title.
  - Tag list: [`[FDConfig]`,`[APIServer]`,`[Engine]`, `[Scheduler]`, `[PD Disaggregation]`, `[Executor]`, `[Graph Optimization]`, `[Speculative Decoding]`, `[RL]`, `[Models]`, `[Quantization]`, `[Loader]`, `[OP]`, `[KVCache]`, `[DataProcessor]`, `[BugFix]`, `[Docs]`, `[CI]`, `[Optimization]`, `[Feature]`, `[Benchmark]`, `[Others]`, `[XPU]`, `[HPU]`, `[GCU]`, `[DCU]`, `[Iluvatar]`, `[Metax]`]
  - You can add new tags based on the PR content, but the semantics must be clear.
- [x] Format your code, run `pre-commit` before commit.
- [x] Add unit tests. Please write the reason in this PR if no unit tests.
- [x] Provide accuracy results.
- [ ] If the current PR is submitting to the `release` branch, make sure the PR has been submitted to the `develop` branch, then cherry-pick it to the `release` branch with the `[Cherry-Pick]` PR tag.

---

问题

级别	文件	概述
🟡 建议	fastdeploy/cache_manager/v1/cache_manager.py:282	[Debug] 前缀的日志使用 logger.info 级别，热路径中每请求都会输出 block_id/ref_count 列表，生产环境将造成严重日志泛洪
🟡 建议	fastdeploy/cache_manager/v1/cache_manager.py:1199	abort_prefetch_blocks 直接调用 _radix_tree._remove_node_from_tree 私有方法，破坏封装
🟡 建议	fastdeploy/cache_manager/v1/cache_controller.py:116	storage_enabled 通过 getattr 读取 _transfer_manager._storage_connector 私有属性，应在 CacheTransferManager 上暴露公共 API
❓ 疑问	fastdeploy/cache_manager/v1/storage/__init__.py:73	create_storage_scheduler 连接失败从"warn and continue"改为抛 RuntimeError，Mooncake 暂时不可用时将导致整个服务无法启动，是否有重试或降级策略？
📝 PR 规范	—	标题含两个 Tag [KVCache][Feature]（规范要求仅一个）；PR body 缺少 ## Accuracy Tests 段落

---

总体评价

整体实现架构清晰，D2H→Storage 链式驱逐的 on_success 回调设计合理；存储层 key 格式统一到 cache_utils.storage_key_for_block 也是良好重构。主要需关注调试日志级别问题（会在生产环境产生大量噪声）和跨类私有属性访问的封装问题，其余为设计层面的讨论点。

[PaddlePaddle-bot]

🟡 建议 多处 [Debug] 前缀的消息使用 logger.info 级别输出（此处及 L321、L529、L558、L775、L1030、L1109、L1165 等）。

每个请求都会打印包含完整 block_id 列表和 ref_count 的 INFO 日志，在高并发生产环境将产生严重日志泛洪，同时 dict/list 的字符串化本身有 CPU 开销。

建议统一改为 logger.debug，或在合并前移除这些临时调试日志：

# 改为
logger.debug(
    f"[allocate_device_blocks] swap host->device: "
    f"host_refs_before={host_refs_before}, host_refs_after={host_refs_after}, ..."
)

[PaddlePaddle-bot]

🟡 建议 abort_prefetch_blocks 直接调用了 RadixTree 的私有方法 _radix_tree._remove_node_from_tree(node)，破坏封装边界。若 RadixTree 内部实现变更，此处会悄然失效。

建议在 RadixTree 上暴露一个公共方法，例如：

# radix_tree.py
def remove_node(self, node: BlockNode) -> None:
    """Remove a single node from the tree and its parent's children map."""
    self._remove_node_from_tree(node)

然后在 CacheManager 中调用 self._radix_tree.remove_node(node)。

[PaddlePaddle-bot]

🟡 建议 storage_enabled 通过 getattr(self._transfer_manager, "_storage_connector", None) 读取另一个类的私有属性，是跨类私有属性访问反模式。

建议在 CacheTransferManager 上添加公共属性：

# transfer_manager.py
@property
def has_storage_connector(self) -> bool:
    return self._storage_connector is not None

然后 CacheController.storage_enabled 改为：

return self._transfer_manager.has_storage_connector

[PaddlePaddle-bot]

❓ 疑问 create_storage_scheduler 的连接失败处理从原来的「warn and continue（返回 scheduler）」改成了抛 RuntimeError。

这意味着若 Mooncake 服务在节点启动时暂时不可用（如网络抖动、服务未就绪），整个推理服务将无法启动。

请确认这是预期行为。如果是，建议在错误信息里补充排查建议（如检查 Mooncake 服务是否已就绪、MOONCAKE_METADATA_SERVER 配置是否正确）；如果希望支持降级运行（无 storage 仍可服务），则应恢复 warn-and-continue 语义。