[KVCache] unify host block allocation through allocate_host_blocks

## Motivation

直接调用 `_host_pool.allocate()` 时不会触发驱逐，导致在 host block 空闲不足
但存在 evictable block 的情况下，`can_allocate_host_blocks` 返回 True 但分配
静默失败。

## Modifications

- `prepare_prefetch_metadata`：将 `_host_pool.allocate()` 替换为 `allocate_host_blocks()`，
  空闲不足时自动驱逐 evictable host block 后再分配
- 删除未被生产代码调用的 `offload_to_host` 方法及其全部测试用例

Author: kevincheng2

fastdeploy/cache_manager/v1/cache_manager.py

@@ -909,45 +909,6 @@ def check_and_add_pending_backup(
         except Exception as e:
             logger.error(f"check_and_add_pending_backup error: {e}, {str(traceback.format_exc())}")
 
-    # ============ Host/Device Transfer Coordination ============
-
-    def offload_to_host(self, block_indices: List[int]) -> bool:
-        """
-        Offload blocks from device to host memory.
-
-        This is a coordination method. Actual data transfer happens in Worker.
-
-        Args:
-            block_indices: List of block indices to offload
-
-        Returns:
-            True if successful, False otherwise
-        """
-        try:
-            with self._lock:
-                # Allocate host blocks
-                host_indices = self._host_pool.allocate(len(block_indices))
-                if host_indices is None or len(host_indices) != len(block_indices):
-                    # Not enough host memory, release what we allocated
-                    if host_indices:
-                        self._host_pool.release(host_indices)
-                    return False
-
-                # Perform the offload (actual data transfer would happen in Worker)
-                for i, dev_idx in enumerate(block_indices):
-                    host_idx = host_indices[i]
-                    metadata = self._device_pool.get_metadata(dev_idx)
-                    if metadata:
-                        self._host_pool.set_metadata(host_idx, metadata)
-
-                # Release device blocks
-                self._device_pool.release(block_indices)
-
-                return True
-        except Exception as e:
-            logger.error(f"offload_to_host error: {e}, {str(traceback.format_exc())}")
-            return False
-
     def load_from_host(self, block_indices: List[int]) -> bool:
         """
         Load blocks from host to device memory.
@@ -1063,9 +1024,9 @@ def prepare_prefetch_metadata(
                 if not self.can_allocate_host_blocks(len(storage_hashes)):
                     return []
 
-                # Allocate host blocks for prefetch
-                host_block_ids = self._host_pool.allocate(len(storage_hashes))
-                if host_block_ids is None or len(host_block_ids) == 0:
+                # Allocate host blocks for prefetch (evicts evictable host blocks if needed)
+                host_block_ids = self.allocate_host_blocks(len(storage_hashes))
+                if not host_block_ids:
                     return []
 
                 blocks = list(zip(storage_hashes, host_block_ids))

tests/cache_manager/v1/test_cache_manager.py

@@ -788,65 +788,6 @@ def test_prefetch_node_map_initially_empty(self):
         self.assertEqual(len(cache_manager._prefetch_node_map), 0)
 
 
-class TestCacheManagerOffloadToHost(unittest.TestCase):
-    """Tests for CacheManager.offload_to_host."""
-
-    def test_offload_frees_device_blocks(self):
-        """After offload, device blocks should be released."""
-        cm = create_cache_manager(total_block_num=20, num_cpu_blocks=20)
-        device_blocks = cm._device_pool.allocate(4)
-        self.assertIsNotNone(device_blocks)
-        free_before = cm.num_free_device_blocks
-
-        success = cm.offload_to_host(device_blocks)
-
-        self.assertTrue(success)
-        self.assertEqual(cm.num_free_device_blocks, free_before + 4)
-
-    def test_offload_allocates_host_blocks(self):
-        """After offload, host blocks should be consumed."""
-        cm = create_cache_manager(total_block_num=20, num_cpu_blocks=20)
-        device_blocks = cm._device_pool.allocate(3)
-        free_host_before = cm.num_free_host_blocks
-
-        cm.offload_to_host(device_blocks)
-
-        self.assertEqual(cm.num_free_host_blocks, free_host_before - 3)
-
-    def test_offload_fails_when_no_host_blocks(self):
-        """Offload should return False when host pool is exhausted."""
-        cm = create_cache_manager(total_block_num=20, num_cpu_blocks=0)
-        device_blocks = cm._device_pool.allocate(2)
-
-        success = cm.offload_to_host(device_blocks)
-        self.assertFalse(success)
-
-    def test_offload_copies_device_metadata_to_host(self):
-        """Metadata on device blocks should be copied to host blocks."""
-        from fastdeploy.cache_manager.v1.metadata import CacheBlockMetadata
-
-        cm = create_cache_manager(total_block_num=20, num_cpu_blocks=20)
-        device_blocks = cm._device_pool.allocate(1)
-        block_id = device_blocks[0]
-        meta = CacheBlockMetadata(block_id=block_id, device_id=0, block_size=64, ref_count=5)
-        cm._device_pool.set_metadata(block_id, meta)
-
-        cm.offload_to_host(device_blocks)
-
-        # Find the newly used host block (last used)
-        used_host = list(cm._host_pool._used_blocks)
-        self.assertEqual(len(used_host), 1)
-        host_meta = cm._host_pool.get_metadata(used_host[0])
-        self.assertIsNotNone(host_meta)
-        self.assertEqual(host_meta.ref_count, 5)
-
-    def test_offload_empty_list_returns_true(self):
-        """Offloading empty list succeeds."""
-        cm = create_cache_manager()
-        success = cm.offload_to_host([])
-        self.assertTrue(success)
-
-
 # ---------------------------------------------------------------------------
 # load_from_host
 # ---------------------------------------------------------------------------