[KVCache][Engine][OP] cache manager v1 refactor, engine fixes and new unit tests

## Motivation

cache_manager_storage_transfer 分支持续迭代，本次提交包含多个改进方向：
1. BlockPool 分配逻辑有性能问题且缺少边界处理
2. CacheController.free_cache 无条件清除 storage 可能误清数据
3. request.py 的 CacheSwapMetadata 使用字符串表示 CacheLevel，类型不安全
4. common_engine / resource_manager / worker 中多模态判断入口不统一
5. radix_tree 中 select_blocks_for_backup 已无使用方，可清理
6. gpu_model_runner 多处可优化：异步写入、新注意力后端、MLA/DSA slot_mapping
7. MACA 平台 ops.py 中 F811 重定义警告
8. 缺少 cache manager v1 核心逻辑的单元测试

## Modifications

- cache_manager/ops.py: 移除 MACA 平台对 swap_cache_per_layer/async 的无效 import（F811）
- cache_manager/v1/block_pool.py: allocate 增加 num_blocks==0 提前返回；批量切片替换循环 pop
- cache_manager/v1/cache_controller.py: free_cache 新增 clear_storage 参数，默认 False
- cache_manager/v1/radix_tree.py: 删除废弃的 select_blocks_for_backup 方法
- engine/request.py: CacheSwapMetadata src/dst_type 改用 CacheLevel 枚举；日志接口统一
- engine/common_engine.py: enable_mm 判断统一为 enable_mm_runtime；V1 cache pause 走
  reset_cache；新增 model_loader_extra_config / enable_flashinfer_allreduce_fusion 透传；
  修复 is_end 时残余 token_ids 未返回的边界 case
- engine/sched/resource_manager_v1.py: enable_mm 统一；block 释放去掉冗余条件分支
- model_executor/forward_meta.py: ForwardMeta 用 slot_mapping 替换 mask_encoder_batch；
  XPUForwardMeta 新增 is_speculative 字段
- worker/gpu_model_runner.py: 引入 DSA/MLA 注意力后端；image processor 通用化；
  share_inputs 改为 async_set_value；新增 _compute_position_ids_and_slot_mapping；
  TBO 双缓冲支持；修复拼写错误；去掉 num_blocks > 40000 硬限制
- worker/worker_process.py: enable_mm_runtime 统一；参数顺序整理
- tests/cache_manager/v1/: 新增 test_cache_manager / test_cache_utils / test_radix_tree
  单元测试，覆盖 offload/load、pending backup、hash、LayerDoneCounter、complete_swap 等

## Usage or Command

```bash
# 运行新增单元测试
source .venv/py310/bin/activate
python -m pytest tests/cache_manager/v1/ -vv -s

# 启动服务（单机）
cd baidu/FastDeploy
bash run.sh
```

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: kevincheng2

fastdeploy/cache_manager/ops.py

@@ -49,12 +49,6 @@ def get_peer_mem_addr(*args, **kwargs):
             raise RuntimeError("CUDA no need of get_peer_mem_addr!")
 
     elif current_platform.is_maca():
-        from fastdeploy.model_executor.ops.gpu import (
-            swap_cache_per_layer,  # 单层 KV cache 换入算子（同步）
-        )
-        from fastdeploy.model_executor.ops.gpu import (
-            swap_cache_per_layer_async,  # 单层 KV cache 换入算子（异步，无强制 sync）
-        )
         from fastdeploy.model_executor.ops.gpu import (  # get_output_kv_signal,; ipc_sent_key_value_cache_by_remote_ptr_block_sync,
             cuda_host_alloc,
             cuda_host_free,

fastdeploy/cache_manager/v1/block_pool.py

@@ -65,18 +65,19 @@ def allocate(self, num_blocks: int) -> Optional[List[int]]:
             List of allocated block indices if successful, None if not enough blocks
         """
         with self._lock:
+            if num_blocks == 0:
+                return []
+
             if num_blocks > len(self._free_blocks):
                 logger.warning(
                     f"BlockPool.allocate failed: not enough blocks, "
                     f"requested={num_blocks}, available={len(self._free_blocks)}"
                 )
                 return None
 
-            allocated = []
-            for _ in range(num_blocks):
-                block_idx = self._free_blocks.pop(0)
-                self._used_blocks.add(block_idx)
-                allocated.append(block_idx)
+            allocated = self._free_blocks[-num_blocks:]
+            del self._free_blocks[-num_blocks:]
+            self._used_blocks.update(allocated)
 
             return allocated
 

fastdeploy/cache_manager/v1/cache_controller.py

@@ -1077,7 +1077,7 @@ def reset_cache(self) -> bool:
         except Exception:
             return False
 
-    def free_cache(self) -> bool:
+    def free_cache(self, clear_storage: bool = False) -> bool:
         """
         Free all cache storage (GPU memory + CPU pinned memory + storage).
 
@@ -1098,7 +1098,8 @@ def free_cache(self) -> bool:
             self._free_host_cache()
 
             # Clear storage
-            self._clear_storage()
+            if clear_storage:
+                self._clear_storage()
 
             return True
         except Exception:

fastdeploy/cache_manager/v1/radix_tree.py

@@ -590,40 +590,6 @@ def complete_swap_to_device(
 
         return gpu_block_ids
 
-    def select_blocks_for_backup(
-        self,
-        needed_num: int,
-    ) -> List[BlockNode]:
-        """
-        Select blocks to backup from evictable device nodes.
-
-        Selects the coldest blocks (LRU) from _evictable_device that don't
-        already have a backup.
-
-        Args:
-            needed_num: Number of blocks to select for backup
-
-        Returns:
-            List of BlockNode objects to backup
-        """
-        if needed_num <= 0:
-            return []
-
-        with self._lock:
-            # Find candidates: evictable device nodes without backup
-            candidates = []
-            for node_id, (_, node) in self._evictable_device.items():
-                if not node.backuped:
-                    candidates.append(node)
-
-            if not candidates:
-                return []
-
-            # Sort by last_access_time (LRU - oldest first)
-            candidates.sort(key=lambda n: n.last_access_time)
-
-            return candidates[:needed_num]
-
     def backup_blocks(
         self,
         nodes: List[BlockNode],

fastdeploy/engine/common_engine.py

@@ -342,6 +342,7 @@ def create_data_processor(self):
             self.cfg.limit_mm_per_prompt,
             self.cfg.mm_processor_kwargs,
             self.cfg.tool_parser,
+            enable_mm_runtime=self.cfg.enable_mm_runtime,
         )
         self.data_processor = self.input_processor.create_processor()
         self.mm_max_tokens_per_item = self.data_processor.get_mm_max_tokens_per_item(
@@ -611,7 +612,7 @@ def insert_tasks(self, tasks: List[Request], current_id=-1):
                         LoggingEventName.RESCHEDULED_INFERENCE_START, task.request_id, getattr(task, "user", "")
                     )
             if not is_prefill:
-                if not self.cfg.model_config.enable_mm:
+                if not self.cfg.enable_mm_runtime:
                     self.update_requests_chunk_size(tasks)
                 else:
                     self.update_mm_requests_chunk_size(tasks)
@@ -1260,7 +1261,7 @@ def _insert_zmq_task_to_scheduler(self):
         while self.running:
             try:
                 block = True if len(added_requests) == 0 else False
-                if not self.cfg.model_config.enable_mm:
+                if not self.cfg.enable_mm_runtime:
                     err, data = self.recv_request_server.receive_json_once(block)
                 else:
                     err, data = self.recv_request_server.receive_pyobj_once(block)
@@ -1482,22 +1483,25 @@ def _control_pause(self, control_request: ControlRequest):
             self._send_error_response(req.request_id, "Request is aborted since engine is paused.")
         self.scheduler.reset()
 
-        # pause cache transfer
-        if self.cfg.cache_config.num_cpu_blocks > 0 or self.cfg.cache_config.kvcache_storage_backend:
-            self.llm_logger.info("Start to pause cache transfer.")
-            pause_transfer_request = ControlRequest(
-                request_id=f"{control_request.request_id}_pause_transfer", method="pause"
-            )
-            self.cache_task_queue.put_transfer_task((CacheStatus.CTRL, pause_transfer_request))
-            # Wait for cache_transfer responses
-            asyncio.run(
-                self._wait_for_control_responses(
-                    f"{pause_transfer_request.request_id}", 60, executors=["cache_transfer"]
+        if envs.ENABLE_V1_KVCACHE_MANAGER:
+            self.resource_manager.cache_manager.reset_cache()
+        else:
+            # pause cache transfer
+            if self.cfg.cache_config.num_cpu_blocks > 0 or self.cfg.cache_config.kvcache_storage_backend:
+                self.llm_logger.info("Start to pause cache transfer.")
+                pause_transfer_request = ControlRequest(
+                    request_id=f"{control_request.request_id}_pause_transfer", method="pause"
                 )
-            )
-            self.llm_logger.info("Successfully paused cache transfer.")
+                self.cache_task_queue.put_transfer_task((CacheStatus.CTRL, pause_transfer_request))
+                # Wait for cache_transfer responses
+                asyncio.run(
+                    self._wait_for_control_responses(
+                        f"{pause_transfer_request.request_id}", 60, executors=["cache_transfer"]
+                    )
+                )
+                self.llm_logger.info("Successfully paused cache transfer.")
 
-        self.resource_manager.cache_manager.reset()
+            self.resource_manager.cache_manager.reset()
         self.llm_logger.info("Successfully paused request generation.")
         return None
 
@@ -1791,10 +1795,14 @@ def _control_sleep(self, control_request: ControlRequest):
             executors.add("worker")
         if "kv_cache" in tags:
             executors.add("worker")
-            if self.cfg.cache_config.num_cpu_blocks > 0 or self.cfg.cache_config.kvcache_storage_backend:
-                executors.add("cache_transfer")
-            if self.cfg.cache_config.enable_prefix_caching:
-                self.resource_manager.cache_manager.reset()
+            if envs.ENABLE_V1_KVCACHE_MANAGER:
+                if self.cfg.cache_config.enable_prefix_caching:
+                    self.resource_manager.cache_manager.reset_cache()
+            else:
+                if self.cfg.cache_config.num_cpu_blocks > 0 or self.cfg.cache_config.kvcache_storage_backend:
+                    executors.add("cache_transfer")
+                if self.cfg.cache_config.enable_prefix_caching:
+                    self.resource_manager.cache_manager.reset()
 
         # Dispatch sleep request to executors
         self.llm_logger.info(f"Dispatch sleep request to executors: {list(executors)}")
@@ -1989,6 +1997,11 @@ def _decode_token(self, token_ids, req_id, is_end):
                 token_ids = cum_tokens[prefix_offset:read_offset]
             else:
                 token_ids = []
+
+            if is_end and delta_text == "" and len(cum_tokens) > 0:
+                read_offset = self.data_processor.decode_status[req_id][1]
+                token_ids = cum_tokens[read_offset:]
+
             if is_end:
                 del self.data_processor.decode_status[req_id]
         return delta_text, token_ids
@@ -2444,7 +2457,7 @@ def _setting_environ_variables(self):
             if self.cfg.scheduler_config.splitwise_role == "prefill":
                 variables["FLAGS_fmt_write_cache_completed_signal"] = 1
 
-        if self.cfg.model_config.enable_mm:
+        if self.cfg.enable_mm_runtime:
             variables["FLAGS_max_partition_size"] = 1024
 
         command_prefix = ""
@@ -2545,6 +2558,7 @@ def _start_worker_service(self):
             f" --early_stop_config '{self.cfg.early_stop_config.to_json_string()}'"
             f" --reasoning_parser {self.cfg.structured_outputs_config.reasoning_parser}"
             f" --load_choices {self.cfg.load_config.load_choices}"
+            f" --model_loader_extra_config '{json.dumps(self.cfg.load_config.model_loader_extra_config)}'"
             f" --plas_attention_config '{self.cfg.plas_attention_config.to_json_string()}'"
             f" --ips {ips}"
             f" --cache-transfer-protocol {self.cfg.cache_config.cache_transfer_protocol}"
@@ -2577,6 +2591,7 @@ def _start_worker_service(self):
             "moe_gate_fp32": self.cfg.model_config.moe_gate_fp32,
             "enable_entropy": self.cfg.model_config.enable_entropy,
             "enable_overlap_schedule": self.cfg.scheduler_config.enable_overlap_schedule,
+            "enable_flashinfer_allreduce_fusion": self.cfg.parallel_config.enable_flashinfer_allreduce_fusion,
         }
         for worker_flag, value in worker_store_true_flag.items():
             if value:

fastdeploy/engine/request.py

@@ -34,7 +34,11 @@
 from typing_extensions import TypeVar
 
 from fastdeploy import envs
-from fastdeploy.cache_manager.v1.metadata import CacheSwapMetadata, PendingPrefetch
+from fastdeploy.cache_manager.v1.metadata import (
+    CacheLevel,
+    CacheSwapMetadata,
+    PendingPrefetch,
+)
 from fastdeploy.engine.pooling_params import PoolingParams
 from fastdeploy.engine.sampling_params import SamplingParams
 from fastdeploy.entrypoints.openai.protocol import (
@@ -43,7 +47,11 @@
     StructuralTagResponseFormat,
     ToolCall,
 )
-from fastdeploy.utils import data_processor_logger
+from fastdeploy.logger.request_logger import (
+    RequestLogLevel,
+    log_request,
+    log_request_error,
+)
 from fastdeploy.worker.output import (
     LogprobsLists,
     PromptLogprobs,
@@ -250,7 +258,7 @@ def prompt_hashes(self) -> list[str]:
         return self._prompt_hashes
 
     @property
-    def match_result(self) -> MatchResult:
+    def match_result(self) -> Optional[MatchResult]:
         return self._match_result
 
     @match_result.setter
@@ -364,15 +372,13 @@ def from_generic_request(
             ), "The parameter `raw_request` is not supported now, please use completion api instead."
             for key, value in req.metadata.items():
                 setattr(request, key, value)
-            from fastdeploy.utils import api_server_logger
-
-            api_server_logger.warning("The parameter metadata is obsolete.")
+            log_request(RequestLogLevel.STAGES, message="The parameter metadata is obsolete.")
 
         return request
 
     @classmethod
     def from_dict(cls, d: dict):
-        data_processor_logger.debug(f"{d}")
+        log_request(RequestLogLevel.FULL, message="{request}", request=d)
         sampling_params: SamplingParams = None
         pooling_params: PoolingParams = None
         metrics: RequestMetrics = None
@@ -403,8 +409,11 @@ def from_dict(cls, d: dict):
                         ImagePosition(**mm_pos) if not isinstance(mm_pos, ImagePosition) else mm_pos
                     )
             except Exception as e:
-                data_processor_logger.error(
-                    f"Convert mm_positions to ImagePosition error: {e}, {str(traceback.format_exc())}"
+                log_request_error(
+                    message="request[{request_id}] Convert mm_positions to ImagePosition error: {error}, {traceback}",
+                    request_id=d.get("request_id"),
+                    error=str(e),
+                    traceback=traceback.format_exc(),
                 )
         return cls(
             request_id=d["request_id"],
@@ -640,8 +649,8 @@ def append_swap_metadata(self, metadata: List[CacheSwapMetadata]):
                 self.cache_swap_metadata = CacheSwapMetadata(
                     src_block_ids=meta.src_block_ids,
                     dst_block_ids=meta.dst_block_ids,
-                    src_type="host",
-                    dst_type="device",
+                    src_type=CacheLevel.HOST,
+                    dst_type=CacheLevel.DEVICE,
                     hash_values=meta.hash_values,
                 )
 
@@ -655,8 +664,8 @@ def append_evict_metadata(self, metadata: List[CacheSwapMetadata]):
                 self.cache_evict_metadata = CacheSwapMetadata(
                     src_block_ids=meta.src_block_ids,
                     dst_block_ids=meta.dst_block_ids,
-                    src_type="device",
-                    dst_type="host",
+                    src_type=CacheLevel.HOST,
+                    dst_type=CacheLevel.DEVICE,
                     hash_values=meta.hash_values,
                 )
 

fastdeploy/engine/sched/resource_manager_v1.py

@@ -245,11 +245,11 @@ def __init__(self, max_num_seqs, config, tensor_parallel_size, splitwise_role, l
         self.need_block_num_map = dict()
 
         self.encoder_cache = None
-        if config.model_config.enable_mm and config.cache_config.max_encoder_cache > 0:
+        if config.enable_mm_runtime and config.cache_config.max_encoder_cache > 0:
             self.encoder_cache = EncoderCacheManager(config.cache_config.max_encoder_cache)
 
         self.processor_cache = None
-        if config.model_config.enable_mm and config.cache_config.max_processor_cache > 0:
+        if config.enable_mm_runtime and config.cache_config.max_processor_cache > 0:
             max_processor_cache_in_bytes = int(config.cache_config.max_processor_cache * 1024 * 1024 * 1024)
             self.processor_cache = ProcessorCacheManager(max_processor_cache_in_bytes)
 
@@ -714,7 +714,7 @@ def _get_num_new_tokens(self, request, token_budget):
             num_new_tokens = token_budget // self.config.cache_config.block_size * self.config.cache_config.block_size
         request.with_image = False
 
-        if not self.config.model_config.enable_mm:
+        if not self.config.enable_mm_runtime:
             return num_new_tokens
 
         inputs = request.multimodal_inputs
@@ -1948,13 +1948,7 @@ def _free_blocks(self, request: Request):
                 request.block_tables[request.num_cached_blocks :], request.request_id
             )
         else:
-            if self.config.cache_config.enable_prefix_caching:
-                self.cache_manager.release_block_ids(request)
-                self.cache_manager.recycle_gpu_blocks(
-                    request.block_tables[request.num_cached_blocks :], request.request_id
-                )
-            else:
-                self.cache_manager.recycle_gpu_blocks(request.block_tables, request.request_id)
+            self.cache_manager.recycle_gpu_blocks(request.block_tables, request.request_id)
         request.block_tables = []
 
         if request.request_id in self.using_extend_tables_req_id:

fastdeploy/model_executor/forward_meta.py

@@ -164,7 +164,8 @@ class ForwardMeta:
 
     # for mla & dsa
     position_ids: Optional[paddle.Tensor] = None
-    mask_encoder_batch: Optional[paddle.Tensor] = None
+    # for kvcache slot
+    slot_mapping: Optional[paddle.Tensor] = None
 
     real_bsz: int = 0
 
@@ -279,6 +280,7 @@ class XPUForwardMeta(ForwardMeta):
     hidden_states: Optional[paddle.Tensor] = None
 
     is_draft: bool = False
+    is_speculative: bool = False
     # max bs
     max_num_seqs: int = 0