[TE] change ascend direct transport docs & fix async transfer disconnect bug
This commit is contained in:
ascend-direct-dev
GitHub
commit
3e67c9e443
|
|
@ -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)`).
|
||||
|
|
|
|||
|
|
@ -30,6 +30,33 @@ Transfer Engine RPC using <协议> listening on <IP>:<实际端口>，记录目
|
|||
./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)`。
|
||||
|
||||
|
|
|
|||
|
|
@ -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 *> &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);
|
||||
|
|
|
|||
|
|
@ -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<adxl::AdxlEngine>();
|
||||
if (!adxl_) return ERR_MEMORY;
|
||||
std::map<adxl::AscendString, adxl::AscendString> 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<adxl::TransferReq>(
|
||||
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;
|
||||
|
|
|
|||
Loading…
Reference in New Issue