[feat] Make ds_max_workers configurable in YuanrongStorageClient

docs/storage_backends/openyuanrong_datasystem.md

@@ -134,6 +134,7 @@ backend:
     worker_port: 31501                 # Port for Yuanrong datasystem worker on each node
     metastore_port: 2379               # Port for metastore service on the head node
     enable_yr_npu_transport: true      # Enable NPU transport for high-performance device-to-device transfer
+    ds_max_workers: 1                  # Worker threads for parallel serialization in the general KV client
     enable_rdma: false                 # Enable host RDMA (H2H) transport via UCX
     ucx_env_vars: {}                   # UCX env vars for dscli subprocess (e.g., {UCX_LOG_FILE: /tmp/ucx.log, UCX_LOG_LEVEL: ERROR})
     worker_args: "--shared_memory_size_mb 8192 --remote_h2d_device_ids 0 --enable_huge_tlb true"
@@ -143,6 +144,7 @@ backend:
 - `auto_init`: Whether to automatically initialize Yuanrong backend. Default is `True`.
 - `worker_port`: Port for Yuanrong datasystem worker on each node.
 - `metastore_port`: Port for metastore service on the head node.
+- `ds_max_workers`: Number of worker threads used to serialize objects in parallel before storing them via the general KV client. Default is `1` (serial packing).
 - `worker_args`: Additional arguments passed to `dscli start` command:
   - `--shared_memory_size_mb`: Shared memory size in MB for datasystem worker.
   - `--enable_huge_tlb`: Configure huge page memory to reduce TLB misses and improve memory access efficiency. Note: may cause system memory shortage, kernel OOM, or system instability.  **Please allocate huge pages before starting datasystem** - refer to [Huge Page Guide](https://pages.openeuler.openatom.cn/openyuanrong-datasystem/docs/zh-cn/latest/appendix/hugepage_guide.html). Before enabling, OS config required (root privilege): `sysctl -w vm.nr_hugepages=<count>` (each page is 2MB, e.g. 65536 for 128GB) and `ulimit -l unlimited` (allow pinning enough memory for RDMA/Ascend).

scripts/performance_test/perftest_config.yaml

@@ -114,6 +114,9 @@ backend:
     metastore_port: 2379
     # If enable npu transport
     enable_yr_npu_transport: true
+    # Number of worker threads used to serialize objects in parallel before storing
+    # them via the general KV client. 1 disables parallelism (serial packing).
+    ds_max_workers: 1
     # Additional config for yuanrong worker.
     # Recommended options for NPU environments:
     #   --remote_h2d_device_ids       Enable RH2D for efficient cross-node data transfer. Specify NPU device IDs (comma-separated).

transfer_queue/config.yaml

@@ -115,6 +115,9 @@ backend:
     metastore_port: 2379
     # Whether to enable npu transport
     enable_yr_npu_transport: false
+    # Number of worker threads used to serialize objects in parallel before storing
+    # them via the general KV client. 1 disables parallelism (serial packing).
+    ds_max_workers: 1
     # Whether to enable host RDMA (H2H) transport via UCX. Requires RDMA NIC hardware and rdma-core driver.
     # See https://pages.openeuler.openatom.cn/openyuanrong-datasystem/docs/zh-cn/latest/best_practices/best_practices_for_rdma.html
     enable_rdma: false

transfer_queue/storage/clients/yuanrong_client.py

@@ -205,6 +205,11 @@ def __init__(self, config: dict):
         if port is None or not isinstance(port, int):
             raise ValueError("Missing or invalid 'worker_port' in config")
 
+        ds_max_workers = config.get("ds_max_workers", self.DS_MAX_WORKERS)
+        if not isinstance(ds_max_workers, int) or ds_max_workers < 1:
+            raise ValueError(f"Invalid 'ds_max_workers' in config: {ds_max_workers}. Expecting a positive integer")
+        self.DS_MAX_WORKERS = ds_max_workers
+
         logger.info(f"Auto-detecting reachable host for Yuanrong port {port}...")
         host = find_reachable_host(port)
         if host is None: