Fix the bug related to the compatibility with openyuanrong_datasystem interface.

Co-authored-by: liwenlin<liwenlin8@huawei.com>


Co-authored-by: liwenlin<liwenlin8@huawei.com>



# message auto-generated for no-merge-commit merge:
!18 merge fix_ds_zcopy into main

[Fix] Fix the bug related to the compatibility with openyuanrong_datasystem interface.

Created-by: Lexie-7
Commit-by: liwenlin
Merged-by: ascend-robot
Description: # Fixing details

- Modified the interface parameters of `mcreate` and `get_buffers` in the `YuanrongStorageClient` connection.

	The return value of the interface corresponding to the openyuanrong_datasystem does not need to include the `status`. Therefore, when calling the corresponding function in TransferQueue, the return value also needs to be adjusted.

- The waiting time for the `get_buffers` operation has been modified.If the data does not exist, do not wait. If the data exists, then proceed with the normal retrieval.

	The meaning of the `timeout_ms` parameter in `get_buffers` is the waiting time for subscription, that is, how long to wait if the data **does not exist**. If the data already exists, it will be retrieved normally and is not affected by this parameter. 
    However, in TransferQueue, the data retrieved by `get_buffers` should definitely exist,  it has nothing to do with the `timeout_ms` parameter. Even if the data is not immediately retrieved, as long as it is known to exist, it will definitely be retrieved no matter how long it takes.


See merge request: Ascend/TransferQueue!18

Author: ascend-robot

tests/test_yuanrong_storage_manager.py

@@ -78,10 +78,10 @@ def side_effect_mcreate(keys, sizes):
             buffers = [MockBuffer(size) for size in sizes]
             for b in buffers:
                 stored_raw_buffers.append(b.MutableData())
-            return 0, buffers
+            return buffers
 
         storage_client._cpu_ds_client.mcreate.side_effect = side_effect_mcreate
-        storage_client._cpu_ds_client.get_buffers.return_value = (0, stored_raw_buffers)
+        storage_client._cpu_ds_client.get_buffers.return_value = stored_raw_buffers
 
         storage_client.mset_zcopy(
             ["tensor_key", "string_key"], [torch.tensor([1.0, 2.0, 3.0], dtype=torch.float32), "hello yuanrong"]

transfer_queue/storage/clients/yuanrong_client.py

@@ -194,7 +194,7 @@ def mset_zcopy(self, keys: list[str], objs: list[Any]):
         assert self._cpu_ds_client is not None, "CPU DS client is not available"
         items_list = [[memoryview(b) for b in _encoder.encode(obj)] for obj in objs]
         packed_sizes = [calc_packed_size(items) for items in items_list]
-        status, buffers = self._cpu_ds_client.mcreate(keys, packed_sizes)
+        buffers = self._cpu_ds_client.mcreate(keys, packed_sizes)
         tasks = [(target.MutableData(), item) for target, item in zip(buffers, items_list, strict=False)]
         with ThreadPoolExecutor(max_workers=DS_MAX_WORKERS) as executor:
             list(executor.map(lambda p: pack_into(*p), tasks))
@@ -210,7 +210,7 @@ def mget_zcopy(self, keys: list[str]) -> list[Any]:
             list[Any]: List of deserialized objects corresponding to the input keys.
         """
         assert self._cpu_ds_client is not None, "CPU DS client is not available"
-        status, buffers = self._cpu_ds_client.get_buffers(keys, timeout_ms=500)
+        buffers = self._cpu_ds_client.get_buffers(keys)
         return [_decoder.decode(unpack_from(buffer)) if buffer is not None else None for buffer in buffers]
 
     def _batch_put(self, keys: list[str], values: list[Any]):