[BugFix][KVCache] fix cupy device id caching and pickle for _match_result

## Motivation

修复两个 bug：
1. `transfer_manager.py` 中每次调用 `cp.cuda.runtime.getDevice()` 存在隐患，应在初始化时缓存为实例变量，保证后续操作使用一致的设备 ID。
2. `request.py` 的 `__getstate__` 未跳过 `_match_result`，该字段包含 BlockNode 树的父子循环引用，pickle 时会触发 `RecursionError`；同时补充 `__setstate__` 确保 unpickle 后字段恢复为安全默认值。

## Modifications

- `transfer_manager.py`：初始化时调用 `cp.cuda.runtime.getDevice()` 并缓存到 `self._cupy_device_id`，后续 `with cp.cuda.Device(...)` 和日志均使用该缓存值。
- `request.py`：
  - `__getstate__` 中将 `_match_result` 加入跳过集合 `_SKIP_KEYS`，避免循环引用导致 pickle 失败。
  - 新增 `__setstate__`，unpickle 后将 `_block_hasher` 和 `_match_result` 恢复为 `None`。

## Usage or Command

Author: kevincheng2

fastdeploy/cache_manager/v1/transfer_manager.py

@@ -92,12 +92,12 @@ def __init__(
         # They run in parallel without waiting for each other
         # Using cupy to avoid affecting Paddle's internal stream state
         if _HAS_CUPY and paddle.is_compiled_with_cuda():
-            cupy_current_device = cp.cuda.runtime.getDevice()
+            self._cupy_device_id = cp.cuda.runtime.getDevice()
             logger.info(
                 f"[TransferManager] Creating streams: local_rank={self._local_rank}, device_id={self._device_id}, "
-                f"cupy_current_device={cupy_current_device}"
+                f"cupy_device_id={self._cupy_device_id}"
             )
-            with cp.cuda.Device(self._device_id):
+            with cp.cuda.Device(self._cupy_device_id):
                 self._input_stream = cp.cuda.Stream(non_blocking=False)
                 self._output_stream = cp.cuda.Stream(non_blocking=False)
             logger.info(
@@ -447,12 +447,11 @@ def _swap_all_layers_async(
 
         stream = self._output_stream if mode == 0 else self._input_stream
         try:
-            cupy_current_device = cp.cuda.runtime.getDevice()
             logger.debug(
                 f"[TransferManager] _swap_all_layers_async: local_rank={self._local_rank}, device_id={self._device_id}, "
-                f"cupy_current_device={cupy_current_device}, stream_device={stream.device_id}, mode={mode}"
+                f"cupy_device_id={self._cupy_device_id}, stream_device={stream.device_id}, mode={mode}"
             )
-            with cp.cuda.Device(self._device_id):
+            with cp.cuda.Device(self._cupy_device_id):
                 with stream:
                     swap_cache_all_layers(
                         self._device_key_caches,
@@ -534,7 +533,7 @@ def _swap_single_layer_async(
             return False
 
         try:
-            with cp.cuda.Device(self._device_id):
+            with cp.cuda.Device(self._cupy_device_id):
                 with stream:
                     swap_cache_per_layer_async(
                         key_cache,
@@ -640,7 +639,7 @@ def record_input_stream_event(self) -> Any:
         if not _HAS_CUPY or self._input_stream is None:
             return None
         try:
-            with cp.cuda.Device(self._device_id):
+            with cp.cuda.Device(self._cupy_device_id):
                 event = cp.cuda.Event()
                 with self._input_stream:
                     event.record()

fastdeploy/engine/request.py

@@ -453,20 +453,30 @@ def __getstate__(self):
         Custom getstate method for pickle support.
         Handles unpicklable attributes by filtering them from __dict__.
         """
-        # Create a filtered dictionary without problematic attributes
+        # Attributes that cannot or need not be pickled for cross-process transfer.
+        # _block_hasher: closure/callable, not picklable.
+        # _match_result: contains BlockNode tree with parent<->children circular
+        #   references, which causes RecursionError during pickling.
+        # async_process_futures: asyncio futures, not picklable.
+        _SKIP_KEYS = {"_block_hasher", "_match_result"}
         filtered_dict = {}
         for key, value in self.__dict__.items():
-            # Skip attributes that are known to contain unpicklable objects
-            if key == "async_process_futures":
-                filtered_dict[key] = []
-            elif key == "_block_hasher":
-                # Skip _block_hasher (closure function, cannot be pickled)
+            if key in _SKIP_KEYS:
                 continue
+            elif key == "async_process_futures":
+                filtered_dict[key] = []
             else:
                 filtered_dict[key] = value
-
         return filtered_dict
 
+    def __setstate__(self, state):
+        self.__dict__.update(state)
+        # Restore fields that were excluded from pickling with safe defaults.
+        if "_block_hasher" not in self.__dict__:
+            self._block_hasher = None
+        if "_match_result" not in self.__dict__:
+            self._match_result = None
+
     def __eq__(self, other):
         """
         EQ operator.