diff --git a/docs/source/design/transfer-engine/ascend_direct_transport.md b/docs/source/design/transfer-engine/ascend_direct_transport.md index 13a14bb8..4f8ee63b 100644 --- a/docs/source/design/transfer-engine/ascend_direct_transport.md +++ b/docs/source/design/transfer-engine/ascend_direct_transport.md @@ -42,6 +42,33 @@ Complete command format is shown below: ./transfer_engine_ascend_direct_perf --metadata_server=P2PHANDSHAKE --local_server_name=127.0.0.1:12346 --operation=write --device_logicid=1 --mode=initiator --block_size=16384 --batch_size=32 --block_iteration=10 --segment_id=127.0.0.1:real_port ``` +### Environment Variables Configuration + +The following environment variables can be configured to control Ascend Direct Transport behavior: + +| Variable | Description | Default Value | Example | +|----------|-------------|---------------|---------| +| `ASCEND_AUTO_CONNECT` | Enable automatic connection management | 0 (disabled) | `ASCEND_AUTO_CONNECT=1` | +| `ASCEND_ENABLE_USE_FABRIC_MEM` | Enable fabric memory transfer mode in Mooncake Store (A3 only) | 0 (disabled) | `ASCEND_ENABLE_USE_FABRIC_MEM=1` | +| `ASCEND_USE_ASYNC_TRANSFER` | Enable asynchronous transfer mode | 0 (disabled) | `ASCEND_USE_ASYNC_TRANSFER=1` | +| `ASCEND_GLOBAL_RESOURCE_CONFIG` | Global resource configuration | - | `ASCEND_GLOBAL_RESOURCE_CONFIG="{\"fabric_memory.max_capacity\":32}"` | +| `ASCEND_CONNECT_TIMEOUT` | Link establishment timeout in milliseconds | 3000 | `ASCEND_CONNECT_TIMEOUT=5000` | +| `ASCEND_TRANSFER_TIMEOUT` | Data transfer timeout in milliseconds | 3000 | `ASCEND_TRANSFER_TIMEOUT=10000` | +| `ASCEND_THREAD_POOL_SIZE` | Number of worker threads in the transfer thread pool | 8 (1 for buffer pool mode) | `ASCEND_THREAD_POOL_SIZE=16` | +| `ASCEND_USE_SHORT_CONNECTION` | Enable short connection mode (disconnect after each transfer) | 0 (disabled) | `ASCEND_USE_SHORT_CONNECTION=1` | +| `ASCEND_BUFFER_POOL` | Buffer pool configuration for intermediate transfer mode (BUFFER_NUM:BUFFER_SIZE_MB) | "0:0" (disabled) | `ASCEND_BUFFER_POOL=4:8` | +| `ASCEND_BASE_PORT` | Base port for ADXL engine port allocation | 11000 | `ASCEND_BASE_PORT=20000` | +| `HCCL_INTRA_ROCE_ENABLE` | Enable RDMA protocol for intra-node communication | 0 (disabled) | `HCCL_INTRA_ROCE_ENABLE=1` | +| `HCCL_RDMA_TIMEOUT` | RDMA packet retransmission timeout coefficient | - | `HCCL_RDMA_TIMEOUT=14` | +| `HCCL_RDMA_RETRY_CNT` | RDMA packet retransmission count | - | `HCCL_RDMA_RETRY_CNT=7` | + +**Detailed Descriptions:** + +- **ASCEND_AUTO_CONNECT**: Requires CANN 9.0 or later. Default is 0, recommended to enable on supported versions: link can be automatically disconnected when the remote end goes offline abnormally. +- **ASCEND_ENABLE_USE_FABRIC_MEM**: Requires CANN 9.0+ and HDK 26.0+. Recommended when using Mooncake Store on supported A3 platforms: it can significantly improve transmission performance. +- **ASCEND_USE_ASYNC_TRANSFER**: Requires CANN 8.5+. Enables HIXL asynchronous transfer mode, defaults to synchronous mode. +- **ASCEND_GLOBAL_RESOURCE_CONFIG**: Configures HIXL global resources. Refer to HIXL documentation for `OPTION_GLOBAL_RESOURCE_CONFIG` settings. + ### Important Notes 1. **Device Setup Required**: Before calling `TransferEngine.initialize()`, you must set the device (e.g., `torch.npu.set_device(0)`). diff --git a/docs/source/zh_archive/ascend_direct_transport.md b/docs/source/zh_archive/ascend_direct_transport.md index cba448de..50ed7dbc 100644 --- a/docs/source/zh_archive/ascend_direct_transport.md +++ b/docs/source/zh_archive/ascend_direct_transport.md @@ -30,6 +30,33 @@ Transfer Engine RPC using <协议> listening on :<实际端口>，记录目 ./transfer_engine_ascend_direct_perf --metadata_server=P2PHANDSHAKE --local_server_name=127.0.0.1:12346 --operation=write --device_logicid=1 --mode=initiator --block_size=16384 --batch_size=32 --block_iteration=10 --segment_id=127.0.0.1:real_port ``` +### 环境变量配置 + +以下环境变量可用于控制Ascend Direct Transport的行为： + +| 变量名 | 描述 | 默认值 | 示例 | +|--------|--------------------------------------|--------|-----------------------------------------------------------------------| +| `ASCEND_AUTO_CONNECT` | 启用自动连接管理 | 0（禁用） | `ASCEND_AUTO_CONNECT=1` | +| `ASCEND_ENABLE_USE_FABRIC_MEM` | 在Mooncake Store中启用fabric内存传输模式（仅A3） | 0（禁用） | `ASCEND_ENABLE_USE_FABRIC_MEM=1` | +| `ASCEND_USE_ASYNC_TRANSFER` | 启用异步传输模式 | 0（禁用） | `ASCEND_USE_ASYNC_TRANSFER=1` | +| `ASCEND_GLOBAL_RESOURCE_CONFIG` | 全局资源配置 | - | `ASCEND_GLOBAL_RESOURCE_CONFIG="{\"fabric_memory.max_capacity\":32}"` | +| `ASCEND_CONNECT_TIMEOUT` | 链路建链超时时间（毫秒） | 3000 | `ASCEND_CONNECT_TIMEOUT=5000` | +| `ASCEND_TRANSFER_TIMEOUT` | 数据传输超时时间（毫秒） | 3000 | `ASCEND_TRANSFER_TIMEOUT=10000` | +| `ASCEND_THREAD_POOL_SIZE` | 传输线程池的工作线程数 | 8（缓冲池模式下为1） | `ASCEND_THREAD_POOL_SIZE=16` | +| `ASCEND_USE_SHORT_CONNECTION` | 启用短连接模式（每次传输后断开） | 0（禁用） | `ASCEND_USE_SHORT_CONNECTION=1` | +| `ASCEND_BUFFER_POOL` | 中转模式缓冲池配置（BUFFER_NUM:BUFFER_SIZE_MB） | "0:0"（禁用） | `ASCEND_BUFFER_POOL=4:8` | +| `ASCEND_BASE_PORT` | ADXL引擎端口分配的基础端口 | 11000 | `ASCEND_BASE_PORT=20000` | +| `HCCL_INTRA_ROCE_ENABLE` | 启用节点内RDMA通信协议 | 0（禁用） | `HCCL_INTRA_ROCE_ENABLE=1` | +| `HCCL_RDMA_TIMEOUT` | RDMA数据包重传超时时间系数 | - | `HCCL_RDMA_TIMEOUT=14` | +| `HCCL_RDMA_RETRY_CNT` | RDMA数据包重传次数 | - | `HCCL_RDMA_RETRY_CNT=7` | + +详细说明： +ASCEND_AUTO_CONNECT: 需要CANN升级到9.0之后的版本，所以默认值为0，在支持该功能的版本推荐启用：当对端异常下线后可以自动断链。 +ASCEND_ENABLE_USE_FABRIC_MEM：需要CANN升级到9.0之后的版本，HDK升级到26.0之后的版本，在支持该功能的版本使用Mooncake Store时推荐启用：可显著提升传输性能。 +ASCEND_USE_ASYNC_TRANSFER: 需要CANN升级到8.5之后的版本，用于开启Hixl异步传输模式，默认为同步模式。 +ASCEND_GLOBAL_RESOURCE_CONFIG：配置Hixl的全局资源，具体查看hixl的文档关于OPTION_GLOBAL_RESOURCE_CONFIG的配置。 + + ### 注意事项（必看） 1. 调用TransferEngine initialize前需要set device, 比如`torch.npu.set_device(0)`。 diff --git a/mooncake-transfer-engine/include/transport/ascend_transport/ascend_direct_transport/ascend_direct_transport.h b/mooncake-transfer-engine/include/transport/ascend_transport/ascend_direct_transport/ascend_direct_transport.h index 79bb1100..06cf84e2 100644 --- a/mooncake-transfer-engine/include/transport/ascend_transport/ascend_direct_transport/ascend_direct_transport.h +++ b/mooncake-transfer-engine/include/transport/ascend_transport/ascend_direct_transport/ascend_direct_transport.h @@ -77,6 +77,9 @@ class AscendDirectTransport : public Transport { private: int allocateLocalSegmentID(); + static std::string GenAdxlEngineName(const std::string &ip, + const uint64_t port); + void queryThread(); void processSliceList(const std::vector &slice_list); @@ -104,7 +107,6 @@ class AscendDirectTransport : public Transport { uint16_t findAdxlListenPort(); - private: int InitAdxlEngine(); int checkAndConnect(const std::string &target_adxl_engine_name); diff --git a/mooncake-transfer-engine/src/transport/ascend_transport/ascend_direct_transport/ascend_direct_transport.cpp b/mooncake-transfer-engine/src/transport/ascend_transport/ascend_direct_transport/ascend_direct_transport.cpp index 6012c1dd..4a71b929 100644 --- a/mooncake-transfer-engine/src/transport/ascend_transport/ascend_direct_transport/ascend_direct_transport.cpp +++ b/mooncake-transfer-engine/src/transport/ascend_transport/ascend_direct_transport/ascend_direct_transport.cpp @@ -203,7 +203,7 @@ int AscendDirectTransport::install(std::string &local_server_name, int AscendDirectTransport::InitAdxlEngine() { auto local_segment_desc = metadata_->getSegmentDescByID(LOCAL_SEGMENT_ID); std::string host_ip = local_segment_desc->rank_info.hostIp; - uint16_t host_port = local_segment_desc->rank_info.hostPort; + auto host_port = local_segment_desc->rank_info.hostPort; adxl_ = std::make_unique(); if (!adxl_) return ERR_MEMORY; std::map options; @@ -283,9 +283,7 @@ int AscendDirectTransport::InitAdxlEngine() { options["GlobalResourceConfig"] = global_resource_config; LOG(INFO) << "Set GlobalResourceConfig to:" << global_resource_config; } - std::string engine_name_str = - (globalConfig().use_ipv6 ? ("[" + host_ip + "]") : host_ip) + ":" + - std::to_string(host_port); + std::string engine_name_str = GenAdxlEngineName(host_ip, host_port); auto adxl_engine_name = adxl::AscendString(engine_name_str.c_str()); LOG(INFO) << "Set adxl engine name to " << adxl_engine_name.GetString(); auto status = adxl_->Initialize(adxl_engine_name, options); @@ -570,8 +568,7 @@ int AscendDirectTransport::allocateLocalSegmentID() { return FAILED; } local_adxl_engine_name_ = - (globalConfig().use_ipv6 ? ("[" + host_ip + "]") : host_ip) + ":" + - std::to_string(desc->rank_info.hostPort); + GenAdxlEngineName(host_ip, desc->rank_info.hostPort); LOG(INFO) << "AscendDirectTransport set segment desc: host_ip=" << host_ip << ", host_port=" << desc->rank_info.hostPort @@ -666,6 +663,14 @@ uint16_t AscendDirectTransport::findAdxlListenPort() { return 0; } +std::string AscendDirectTransport::GenAdxlEngineName(const std::string &ip, + const uint64_t port) { + if (globalConfig().use_ipv6) { + return ("[" + ip + "]") + ":" + std::to_string(port); + } + return ip + ":" + std::to_string(port); +} + void AscendDirectTransport::queryThread() { #ifdef EXIST_ADXL_ASYNC_METHOD LOG(INFO) << "AscendDirectTransport query thread started"; @@ -699,6 +704,15 @@ void AscendDirectTransport::queryThread() { it = pending_batches.erase(it); continue; } + auto target_segment_desc = + metadata_->getSegmentDescByID(slice_list[0]->target_id); + if (!target_segment_desc) { + it = pending_batches.erase(it); + continue; + } + auto target_adxl_engine_name = + GenAdxlEngineName(target_segment_desc->rank_info.hostIp, + target_segment_desc->rank_info.hostPort); auto handle = static_cast( slice_list[0]->ascend_direct.handle); adxl::TransferStatus task_status; @@ -712,22 +726,12 @@ void AscendDirectTransport::queryThread() { slice->markFailed(); } it = pending_batches.erase(it); + disconnect(target_adxl_engine_name, connect_timeout_); } else if (task_status == adxl::TransferStatus::COMPLETED) { auto now = getCurrentTimeInNano(); auto duration = now - slice_list[0]->ascend_direct.start_time; - auto target_segment_desc = - metadata_->getSegmentDescByID(slice_list[0]->target_id); - if (target_segment_desc) { - auto target_adxl_engine_name = - (globalConfig().use_ipv6 - ? ("[" + target_segment_desc->rank_info.hostIp + - "]") - : target_segment_desc->rank_info.hostIp) + - ":" + - std::to_string(target_segment_desc->rank_info.hostPort); - VLOG(1) << "Transfer to " << target_adxl_engine_name - << " time: " << duration / 1000 << "us"; - } + VLOG(1) << "Transfer to " << target_adxl_engine_name + << " time: " << duration / 1000 << "us"; for (auto &slice : slice_list) { slice->markSuccess(); } @@ -744,6 +748,7 @@ void AscendDirectTransport::queryThread() { for (auto &slice : slice_list) { slice->markFailed(); } + disconnect(target_adxl_engine_name, connect_timeout_); it = pending_batches.erase(it); } else { task_finished = false; @@ -788,10 +793,8 @@ void AscendDirectTransport::processSliceList( need_update_metadata_segs_.erase(it); } auto target_adxl_engine_name = - (globalConfig().use_ipv6 - ? ("[" + target_segment_desc->rank_info.hostIp + "]") - : target_segment_desc->rank_info.hostIp) + - ":" + std::to_string(target_segment_desc->rank_info.hostPort); + GenAdxlEngineName(target_segment_desc->rank_info.hostIp, + target_segment_desc->rank_info.hostPort); adxl::TransferOp operation; if (slice_list[0]->opcode == TransferRequest::WRITE) { operation = adxl::WRITE;