diff --git a/docs/storage_backends/openyuanrong_datasystem.md b/docs/storage_backends/openyuanrong_datasystem.md index 37c7a216..0bf75ef2 100644 --- a/docs/storage_backends/openyuanrong_datasystem.md +++ b/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=` (each page is 2MB, e.g. 65536 for 128GB) and `ulimit -l unlimited` (allow pinning enough memory for RDMA/Ascend). diff --git a/scripts/performance_test/perftest_config.yaml b/scripts/performance_test/perftest_config.yaml index a01a566c..bf2692fa 100644 --- a/scripts/performance_test/perftest_config.yaml +++ b/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). diff --git a/transfer_queue/config.yaml b/transfer_queue/config.yaml index 66fb748c..bd83a599 100644 --- a/transfer_queue/config.yaml +++ b/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 diff --git a/transfer_queue/storage/clients/yuanrong_client.py b/transfer_queue/storage/clients/yuanrong_client.py index 6dc5c92a..5ad54576 100644 --- a/transfer_queue/storage/clients/yuanrong_client.py +++ b/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: