feat: refactor storage prefetch - 3-phase architecture

Refactor _prefetch_storage_cache into three decoupled phases:
- Phase 1 (preprocess thread): CacheManager.prefetch_storage() does matching + enqueue
- Phase 2 (schedule thread): drain pending list, attach to batch_request for dispatch
- Phase 3 (receiver thread): zmq.Poller receives done msgs, stores results

Worker side: extract prefetch tasks from batch_request, execute via thread pool,
send completion via ZMQ PUSH.

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

Author: kevincheng2

fastdeploy/cache_manager/v1/__init__.py

@@ -25,6 +25,7 @@
     CacheStatus,
     MatchResult,
     PDTransferMetadata,
+    PendingPrefetch,
     StorageConfig,
     StorageMetadata,
     StorageType,
@@ -61,6 +62,7 @@
     "AsyncTaskHandler",
     "MatchResult",
     "StorageMetadata",
+    "PendingPrefetch",
     "PDTransferMetadata",
     "StorageConfig",
     "StorageType",

fastdeploy/cache_manager/v1/cache_manager.py

@@ -30,7 +30,15 @@
 from .base import KVCacheBase
 from .block_pool import DeviceBlockPool, HostBlockPool
 from .cache_utils import storage_key_for_block
-from .metadata import BlockNode, CacheLevel, CacheStatus, CacheSwapMetadata, MatchResult
+from .metadata import (
+    BlockNode,
+    CacheLevel,
+    CacheStatus,
+    CacheSwapMetadata,
+    MatchResult,
+    PendingPrefetch,
+    StorageMetadata,
+)
 from .radix_tree import RadixTree
 from .storage import create_storage_scheduler
 
@@ -111,6 +119,10 @@ def __init__(
         # used to quickly update status to HOST once prefetch completes.
         self._prefetch_node_map: Dict[int, BlockNode] = {}
 
+        # Pending prefetch queue: tasks waiting to be dispatched by scheduler
+        self._pending_prefetch_list: List[PendingPrefetch] = []
+        self._pending_prefetch_lock = threading.Lock()
+
         # Storage scheduler (create using factory method if backend is configured)
         self._storage_scheduler = create_storage_scheduler(self.cache_config)
 
@@ -504,10 +516,11 @@ def match_prefix(
                 #   Split matched_nodes into device blocks and host blocks
                 if self.enable_host_cache:
                     for node in matched_nodes:
-                        if node.is_on_device():
-                            result.device_nodes.append(node)
-                        elif node.is_on_host():
-                            result.host_nodes.append(node)
+                        pass
+                        # if node.is_on_device():
+                        #     result.device_nodes.append(node)
+                        # elif node.is_on_host():
+                        #     result.host_nodes.append(node)
                 else:
                     result.device_nodes = matched_nodes
 
@@ -968,6 +981,61 @@ def load_from_host(self, block_indices: List[int]) -> bool:
 
     # ============ Prefetch Methods ============
 
+    def prefetch_storage(self, request: "Request") -> bool:
+        """
+        Execute storage matching and enqueue prefetch info for later dispatch.
+
+        Called from the preprocess thread. Does match_prefix(skip_storage=False)
+        to probe storage, allocate host blocks, and enqueue PendingPrefetch
+        into the pending list. The scheduler will drain and dispatch later.
+
+        Args:
+            request: The request to prefetch cache for.
+
+        Returns:
+            True if storage blocks were matched and enqueued, False otherwise.
+        """
+        if not self.enable_prefix_caching:
+            return False
+
+        self.match_prefix(request, skip_storage=False)
+        match_result = request.match_result
+        request.match_result = None
+
+        if match_result is None or match_result.matched_storage_nums == 0:
+            return False
+
+        storage_nodes = match_result.storage_nodes
+        host_block_ids = [node.block_id for node in storage_nodes]
+        hash_values = [node.hash_value for node in storage_nodes]
+
+        metadata = StorageMetadata(
+            hash_values=hash_values,
+            block_ids=host_block_ids,
+            direction="load",
+        )
+
+        pending = PendingPrefetch(
+            request_id=request.request_id,
+            metadata=metadata,
+            host_block_ids=host_block_ids,
+        )
+        with self._pending_prefetch_lock:
+            self._pending_prefetch_list.append(pending)
+
+        logger.info(
+            f"[Debug][StoragePrefetch] request_id={request.request_id} "
+            f"storage_matched={match_result.matched_storage_nums} blocks, enqueued for dispatch"
+        )
+        return True
+
+    def drain_pending_prefetches(self) -> List[PendingPrefetch]:
+        """Atomically drain all pending prefetch tasks for scheduler dispatch."""
+        with self._pending_prefetch_lock:
+            items = self._pending_prefetch_list
+            self._pending_prefetch_list = []
+            return items
+
     def prepare_prefetch_metadata(
         self,
         storage_hashes: List[str],

fastdeploy/cache_manager/v1/metadata.py

@@ -409,6 +409,23 @@ class StorageMetadata:
     extra_params: Dict[str, Any] = field(default_factory=dict)
 
 
+@dataclass
+class PendingPrefetch:
+    """
+    Represents a pending storage prefetch task enqueued by CacheManager,
+    waiting to be dispatched to workers by the scheduler.
+
+    Attributes:
+        request_id: The request that triggered this prefetch.
+        metadata: StorageMetadata with hash_values and block_ids for the transfer.
+        host_block_ids: Pre-allocated host block IDs (for cleanup on failure).
+    """
+
+    request_id: str = ""
+    metadata: "StorageMetadata" = field(default_factory=lambda: StorageMetadata())
+    host_block_ids: List[int] = field(default_factory=list)
+
+
 @dataclass
 class PDTransferMetadata:
     """

fastdeploy/engine/common_engine.py

@@ -1168,6 +1168,12 @@ def _fetch_request():
                                 task.metrics.decode_inference_start_time = time.time()
                             elif not task.has_been_preempted_before:
                                 task.metrics.inference_start_time = time.time()
+                    if batch_request.storage_prefetch_tasks:
+                        self.llm_logger.info(
+                            f"[Debug][StoragePrefetch][Dispatch] put_tasks with "
+                            f"{len(batch_request.storage_prefetch_tasks)} prefetch tasks, "
+                            f"{len(batch_request.requests)} inference requests"
+                        )
                     self.engine_worker_queue.put_tasks((batch_request, self.resource_manager.real_bsz))
                 else:
                     # When there are no actual tasks to schedule, send an empty task batch to EP workers.

fastdeploy/engine/request.py

@@ -34,7 +34,7 @@
 from typing_extensions import TypeVar
 
 from fastdeploy import envs
-from fastdeploy.cache_manager.v1.metadata import CacheSwapMetadata
+from fastdeploy.cache_manager.v1.metadata import CacheSwapMetadata, PendingPrefetch
 from fastdeploy.engine.pooling_params import PoolingParams
 from fastdeploy.engine.sampling_params import SamplingParams
 from fastdeploy.entrypoints.openai.protocol import (
@@ -618,6 +618,7 @@ def __init__(self):
 
         self.cache_swap_metadata: Optional[CacheSwapMetadata] = None
         self.cache_evict_metadata: Optional[CacheSwapMetadata] = None
+        self.storage_prefetch_tasks: Optional[List[PendingPrefetch]] = None
 
     def add_request(self, request):
         if hasattr(request, "cache_swap_metadata") and request.cache_swap_metadata:
@@ -659,9 +660,17 @@ def append_evict_metadata(self, metadata: List[CacheSwapMetadata]):
                     hash_values=meta.hash_values,
                 )
 
+    def append_prefetch_tasks(self, tasks: List[PendingPrefetch]):
+        if self.storage_prefetch_tasks is None:
+            self.storage_prefetch_tasks = []
+        self.storage_prefetch_tasks.extend(tasks)
+
     def __repr__(self):
         requests_repr = repr(self.requests)
-        return f"BatchRequest(requests={requests_repr}, swap_metadata={self.cache_swap_metadata}, evict_metadata={self.cache_evict_metadata})"
+        return (
+            f"BatchRequest(requests={requests_repr}, swap_metadata={self.cache_swap_metadata}, "
+            f"evict_metadata={self.cache_evict_metadata}, prefetch_tasks={self.storage_prefetch_tasks})"
+        )
 
     def __getstate__(self):
         state = self.__dict__.copy()
@@ -688,14 +697,19 @@ def __getitem__(self, index):
         return self.requests[index]
 
     def __len__(self):
-        return len(self.requests)
+        count = len(self.requests)
+        if self.storage_prefetch_tasks:
+            count += len(self.storage_prefetch_tasks)
+        return count
 
     def append(self, batch_request: "BatchRequest"):
         self.requests.extend(batch_request.requests)
         if batch_request.cache_swap_metadata:
             self.append_swap_metadata([batch_request.cache_swap_metadata])
         if batch_request.cache_evict_metadata:
             self.append_evict_metadata([batch_request.cache_evict_metadata])
+        if batch_request.storage_prefetch_tasks:
+            self.append_prefetch_tasks(batch_request.storage_prefetch_tasks)
 
     def extend(self, batch_requests: list["BatchRequest"]):
         for br in batch_requests: