[Bugfix] Fixed YuanrongStorageClient

Co-authored-by: dpj135<diaopengjie@huawei.com>



# message auto-generated for no-merge-commit merge:
!6 merge dpj/fix_yr into main

[Bugfix] Fixed YuanrongStorageClient

Created-by: dpj135
Commit-by: dpj135
Merged-by: ascend-robot
Description: ## Description
We fixed three problems about YuanrongStorageClient :
- The import path for datasystem has changed.
- `_batch_put` lacks a logic branch when GPU is unavailable.
- Tensor serialization has been temporarily optimized using `tensor.clone`.


See merge request: Ascend/TransferQueue!6

Author: dpj135

transfer_queue/storage/clients/yuanrong_client.py

@@ -32,7 +32,7 @@
 YUANRONG_DATASYSTEM_IMPORTED: bool = True
 TORCH_NPU_IMPORTED: bool = True
 try:
-    import datasystem
+    from yr import datasystem
 except ImportError:
     YUANRONG_DATASYSTEM_IMPORTED = False
 
@@ -123,15 +123,17 @@ def _batch_put(self, keys: list[str], values: list[Any]):
             cpu_values = []
 
             for key, value in zip(keys, values, strict=True):
-                if isinstance(value, torch.Tensor) and value.device.type == "npu":
+                if isinstance(value, Tensor) and value.device.type == "npu":
                     if not value.is_contiguous():
                         raise ValueError(f"NPU Tensor is not contiguous: {value}")
                     npu_keys.append(key)
                     npu_values.append(value)
 
                 else:
                     cpu_keys.append(key)
-                    cpu_values.append(pickle.dumps(value))
+                    # TODO: Optimize serialization of tensors
+                    # Serializing slice of tensors results in entire tensors being serialized
+                    cpu_values.append(pickle.dumps(value.clone() if isinstance(value, Tensor) else value))
 
             # put NPU data
             for i in range(0, len(npu_keys), NPU_DS_CLIENT_KEYS_LIMIT):
@@ -153,7 +155,7 @@ def _batch_put(self, keys: list[str], values: list[Any]):
 
         else:
             #  All data goes through CPU path
-            pickled_values = [pickle.dumps(v) for v in values]
+            pickled_values = [pickle.dumps(v.clone() if isinstance(v, Tensor) else v) for v in values]
             for i in range(0, len(keys), CPU_DS_CLIENT_KEYS_LIMIT):
                 batch_keys = keys[i : i + CPU_DS_CLIENT_KEYS_LIMIT]
                 batch_vals = pickled_values[i : i + CPU_DS_CLIENT_KEYS_LIMIT]
@@ -248,6 +250,17 @@ def _batch_get(self, keys: list[str], shapes: list, dtypes: list) -> list[Any]:
                     results[idx] = pickle.loads(raw_val)
 
             return results
+        else:
+            # npu is not available, goes through cpu_ds_client
+            results = [None] * len(keys)
+            idx = 0
+            for i in range(0, len(keys), CPU_DS_CLIENT_KEYS_LIMIT):
+                batch_keys = keys[i : i + CPU_DS_CLIENT_KEYS_LIMIT]
+                raw_values = self._cpu_ds_client.get(batch_keys)
+                for raw_val in raw_values:
+                    results[idx] = pickle.loads(raw_val)
+                    idx += 1
+            return results
 
     def get(self, keys: list[str], shapes=None, dtypes=None) -> list[Any]:
         """Retrieves multiple values from remote storage with expected metadata.