Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include <arpa/inet.h>
12 : #include <securec.h>
13 : #include <string>
14 : #include "network/hccp_common.h"
15 : #include "network_manager_pub.h"
16 : #include "externalinput_pub.h"
17 : #include "hccl_network.h"
18 : #include "transport_direct_npu.h"
19 : #include "launch_aicpu.h"
20 : #include "acl/acl_rt.h"
21 :
22 : using namespace std;
23 :
24 : namespace hccl {
25 : UniversalConcurrentMap<u64, TransportDirectNpu*> TransportDirectNpu::g_qpn2IbversLinkMap_;
26 : bool TransportDirectNpu::g_flag = false;
27 : bool TransportDirectNpu::g_isSupCqeErrInfoListConfig = false;
28 : u32 TransportDirectNpu::cqeErrQpn_ = 0;
29 :
30 : constexpr u32 DEV_PHY_ID_BIT = 32;
31 : constexpr u32 CQE_ARRAY_SIZE = 128;
32 :
33 0 : TransportDirectNpu::TransportDirectNpu(
34 : DispatcherPub* dispatcher, const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara,
35 0 : std::chrono::milliseconds timeout)
36 : : TransportNet(dispatcher, notifyPool, machinePara, timeout),
37 0 : qpsPerConnection_(1),
38 0 : access_(RA_ACCESS_LOCAL_WRITE | RA_ACCESS_REMOTE_WRITE | RA_ACCESS_REMOTE_READ),
39 0 : workFlowMode_(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB),
40 0 : currentQP_(0),
41 0 : qpMode_(machinePara.qpMode)
42 0 : {}
43 :
44 0 : TransportDirectNpu::~TransportDirectNpu()
45 : {
46 0 : HCCL_DEBUG("~TransportDirectNpu Enter!");
47 :
48 0 : (void)DeInit();
49 0 : UnloadAICPUKernel();
50 :
51 0 : HCCL_DEBUG("~TransportDirectNpu Success!");
52 0 : }
53 :
54 0 : HcclResult TransportDirectNpu::DeInit()
55 : {
56 0 : (void)DeRegMR();
57 :
58 0 : (void)DestroyQP();
59 :
60 0 : (void)DestroyAicpuMem();
61 :
62 0 : return HCCL_SUCCESS;
63 : }
64 :
65 0 : HcclResult TransportDirectNpu::GetRemoteMem(UserMemType memType, void** remotePtr)
66 : {
67 0 : HCCL_INFO("[TransportDirectNpu][GetRemoteMem] direct npu getRemoteMem");
68 0 : switch (memType) {
69 0 : case UserMemType::INPUT_MEM:
70 : case UserMemType::OUTPUT_MEM:
71 0 : *remotePtr = remoteMemMsg_[static_cast<u32>(memType)].addr;
72 0 : break;
73 :
74 0 : default:
75 0 : HCCL_ERROR("[Get][RemoteMem]not support dst_mem_type=%d", memType);
76 0 : return HCCL_E_NOT_SUPPORT;
77 : }
78 0 : return HCCL_SUCCESS;
79 : }
80 :
81 0 : HcclResult TransportDirectNpu::GetLocalNotifyValueAddrKey(std::vector<AddrKey>& notifyValue)
82 : {
83 : // 目前这里时随意填充了一些数据上去，不然会调用到基类接口，导致报错
84 0 : AddrKey notifyDetails;
85 0 : notifyDetails.addr = reinterpret_cast<u64>(memMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].addr);
86 0 : notifyDetails.key = reinterpret_cast<u32>(memMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].lkey);
87 0 : notifyValue.push_back(notifyDetails);
88 0 : notifyValue.push_back(notifyDetails);
89 0 : notifyValue.push_back(notifyDetails);
90 0 : return HCCL_SUCCESS;
91 : }
92 :
93 0 : HcclResult TransportDirectNpu::GetLocalRdmaNotify(std::vector<HcclSignalInfo>& rdmaNotify)
94 : {
95 0 : HcclSignalInfo signalInfo = {};
96 0 : rdmaNotify.push_back(signalInfo);
97 0 : rdmaNotify.push_back(signalInfo);
98 0 : rdmaNotify.push_back(signalInfo);
99 0 : return HCCL_SUCCESS;
100 : }
101 :
102 0 : HcclResult TransportDirectNpu::GetRemoteRdmaNotifyAddrKey(std::vector<AddrKey>& rdmaNotifyAddr)
103 : {
104 0 : AddrKey notifyDetails;
105 0 : notifyDetails.addr = reinterpret_cast<u64>(memMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].addr);
106 0 : notifyDetails.key = reinterpret_cast<u32>(memMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].lkey);
107 0 : rdmaNotifyAddr.push_back(notifyDetails);
108 0 : rdmaNotifyAddr.push_back(notifyDetails);
109 0 : rdmaNotifyAddr.push_back(notifyDetails);
110 0 : return HCCL_SUCCESS;
111 : }
112 :
113 0 : HcclResult TransportDirectNpu::GetRemoteMemSize(UserMemType memType, u64& size)
114 : {
115 0 : HCCL_INFO("[TransportDirectNpu][GetRemoteMem] direct npu GetRemoteMemSize");
116 0 : switch (memType) {
117 0 : case UserMemType::INPUT_MEM:
118 : case UserMemType::OUTPUT_MEM:
119 0 : size = remoteMemMsg_[static_cast<u32>(memType)].len;
120 0 : break;
121 :
122 0 : default:
123 0 : HCCL_ERROR("[Get][RemoteMem]not support dst_mem_type=%d", memType);
124 0 : return HCCL_E_NOT_SUPPORT;
125 : }
126 0 : return HCCL_SUCCESS;
127 : }
128 :
129 0 : HcclResult TransportDirectNpu::LoadBinaryFromFile(
130 : const char* binPath, aclrtBinaryLoadOptionType optionType, uint32_t cpuKernelMode, aclrtBinHandle& binHandle)
131 : {
132 : #ifndef CCL_KERNEL
133 : CHK_PRT_RET(binPath == nullptr, HCCL_ERROR("[LoadBinaryFromFile] binary path is nullptr"), HCCL_E_PTR);
134 :
135 : char realPath[PATH_MAX] = {0};
136 : CHK_PRT_RET(
137 : realpath(binPath, realPath) == nullptr,
138 : HCCL_ERROR("LoadBinaryFromFile: %s is not a valid real path, err[%d]", binPath, errno), HCCL_E_INTERNAL);
139 : HCCL_INFO("[LoadBinaryFromFile]realPath: %s", realPath);
140 :
141 : aclrtBinaryLoadOptions loadOptions = {0};
142 : aclrtBinaryLoadOption option;
143 : loadOptions.numOpt = 1;
144 : loadOptions.options = &option;
145 : option.type = optionType;
146 : option.value.cpuKernelMode = cpuKernelMode;
147 : aclError aclRet
148 : = aclrtBinaryLoadFromFile(realPath, &loadOptions, &binHandle); // ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE
149 : CHK_PRT_RET(
150 : aclRet != ACL_SUCCESS, HCCL_ERROR("[LoadBinaryFromFile]errNo[0x%016llx] load binary from file error.", aclRet),
151 : HCCL_E_OPEN_FILE_FAILURE);
152 : #else
153 0 : HCCL_ERROR("[AicpuAclKernelLaunch]Does not support this interface.");
154 0 : return HCCL_E_NOT_SUPPORT;
155 : #endif
156 : return HCCL_SUCCESS;
157 : }
158 :
159 0 : HcclResult TransportDirectNpu::LoadAICPUKernel(void)
160 : {
161 : #ifndef CCL_KERNEL
162 : std::string jsonPath;
163 : CHK_RET(GetKernelFilePath(jsonPath));
164 : jsonPath += "ccl_kernel.json";
165 : HcclResult ret = LoadBinaryFromFile(jsonPath.c_str(), ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0, binHandle_);
166 : CHK_PRT_RET(
167 : ret != HCCL_SUCCESS,
168 : HCCL_ERROR(
169 : "[LoadAICPUKernel]errNo[0x%016llx]load aicpu file fail, path[%s] optionType[%u]"
170 : "cpuKernelMode[%u].",
171 : ret, jsonPath.c_str(), ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0),
172 : ret);
173 : #endif
174 0 : return HCCL_SUCCESS;
175 : }
176 :
177 0 : void TransportDirectNpu::UnloadAICPUKernel(void)
178 : {
179 : #ifndef CCL_KERNEL
180 : if (binHandle_ != nullptr) {
181 : aclError aclRet = aclrtBinaryUnLoad(binHandle_);
182 : if (aclRet != ACL_SUCCESS) {
183 : HCCL_ERROR(
184 : "[UnloadAICPUKernel]errNo[0x%016llx] unload binary from binHandel[%p] error.", aclRet, binHandle_);
185 : }
186 : binHandle_ = nullptr;
187 : }
188 : #endif
189 0 : return;
190 : }
191 :
192 0 : HcclResult TransportDirectNpu::DeRegOneMR(QpHandle& qpHandle, MemMsg& memMsg)
193 : {
194 0 : struct MrInfoT mrInfo = {nullptr};
195 0 : mrInfo.addr = memMsg.addr;
196 0 : HcclResult ret = HrtRaMrDereg(qpHandle, &mrInfo);
197 0 : CHK_PRT_RET(
198 : ret != HCCL_SUCCESS,
199 : HCCL_ERROR("errNo[0x%016llx] in link lbv, In lbv exp deconstruct, mr dereg failed.", HCCL_ERROR_CODE(ret)),
200 : ret);
201 0 : return HCCL_SUCCESS;
202 : }
203 :
204 0 : void TransportDirectNpu::DeRegMRForQPhandles(MemMsg& memMsg)
205 : {
206 0 : for (u32 j = 0; j < qpHandles_.size(); j++) {
207 0 : if (qpHandles_[j] == nullptr) {
208 0 : continue;
209 : }
210 0 : (void)DeRegOneMR(qpHandles_[j], memMsg);
211 : }
212 0 : }
213 :
214 0 : HcclResult TransportDirectNpu::DeRegMR()
215 : {
216 : /* 销毁mr */
217 0 : std::map<uintptr_t, s32> addrMap;
218 0 : for (s32 i = 0; i < static_cast<s32>(MemType::MEM_TYPE_RESERVED); i++) {
219 0 : if (memMsg_[i].mrRegFlag == REG_VALID) {
220 : std::pair<std::map<uintptr_t, s32>::iterator, bool> res
221 0 : = addrMap.insert(std::pair<uintptr_t, s32>(reinterpret_cast<uintptr_t>(memMsg_[i].addr), 0));
222 0 : if (res.second) {
223 0 : DeRegMRForQPhandles(memMsg_[i]);
224 : }
225 : }
226 : }
227 0 : return HCCL_SUCCESS;
228 0 : }
229 :
230 0 : HcclResult TransportDirectNpu::DestroyQpVct(std::vector<QpHandle>& qpHandles)
231 : {
232 : HcclResult ret;
233 0 : for (u32 i = 0; i < qpHandles.size(); i++) {
234 0 : if (qpHandles[i] != nullptr) {
235 0 : struct QpAttr attr {};
236 0 : CHK_RET(hrtRaGetQpAttr(qpHandles[i], &attr));
237 :
238 0 : g_qpn2IbversLinkMap_.Erase(((static_cast<u64>(machinePara_.localDeviceId) << DEV_PHY_ID_BIT) | attr.qpn));
239 :
240 0 : ret = HrtRaQpDestroy(qpHandles[i]);
241 0 : if (ret != HCCL_SUCCESS) {
242 0 : HCCL_ERROR(
243 : "errNo[0x%016llx] in link lbv, lbv exp deconstruct, qp destroy failed.", HCCL_ERROR_CODE(ret));
244 : }
245 0 : qpHandles[i] = nullptr;
246 : }
247 : }
248 0 : return HCCL_SUCCESS;
249 : }
250 :
251 0 : HcclResult TransportDirectNpu::DestroyQP()
252 : {
253 0 : CHK_RET(DestroyQpVct(qpHandles_));
254 0 : return HCCL_SUCCESS;
255 : }
256 :
257 0 : HcclResult TransportDirectNpu::DestroyAicpuMem()
258 : {
259 0 : if (aicpuMem_.ptr() != nullptr) {
260 0 : aicpuMem_.free();
261 : }
262 0 : return HCCL_SUCCESS;
263 : }
264 :
265 0 : HcclResult TransportDirectNpu::CreateAicpuMem()
266 : {
267 0 : CHK_RET(DeviceMem::alloc(aicpuMem_, AICPU_FLAG_AREA));
268 :
269 0 : CHK_RET(hrtMemSet(aicpuMem_.ptr(), aicpuMem_.size(), aicpuMem_.size()));
270 :
271 0 : HCCL_INFO("[TransportDirectNpu][CreateAicpuMem] buffer ptr[%p]", aicpuMem_.ptr());
272 0 : return HCCL_SUCCESS;
273 : }
274 :
275 0 : HcclResult TransportDirectNpu::Init()
276 : {
277 0 : HCCL_INFO(
278 : "machineType=[%d], serverId=[%s], localDeviceId=[%d], remoteDeviceId=[%d], "
279 : "localRank=[%u], localUserRank=[%u], remoteRank=[%u], remoteUserrank=[%u], "
280 : "deviceType=[%d], inputMem=[%p], outputMem=[%p], isAicpuModeEn[%d], notifyNum[%u], "
281 : "custom exchange data size [%llu].",
282 : machinePara_.machineType, machinePara_.serverId.c_str(), machinePara_.localDeviceId,
283 : machinePara_.remoteDeviceId, machinePara_.localUserrank, machinePara_.localWorldRank,
284 : machinePara_.remoteUserrank, machinePara_.remoteWorldRank, machinePara_.deviceType, machinePara_.inputMem.ptr(),
285 : machinePara_.outputMem.ptr(), machinePara_.isAicpuModeEn, machinePara_.notifyNum,
286 : machinePara_.exchangeInfo.size());
287 0 : HcclUs startut = TIME_NOW();
288 :
289 0 : CHK_SMART_PTR_NULL(machinePara_.inputMem);
290 0 : CHK_SMART_PTR_NULL(machinePara_.outputMem);
291 0 : CHK_RET(CheckDeviceId());
292 0 : CHK_RET(CheckExchangeData());
293 0 : CHK_RET(CreateAicpuMem());
294 0 : CHK_RET(LoadAICPUKernel());
295 :
296 : // 上层初始化时保证 machinePara_.sockets 非空
297 0 : if (machinePara_.sockets.size() == 0) {
298 0 : HCCL_ERROR("machinePara sockets is empty.");
299 0 : return HCCL_E_INTERNAL;
300 : }
301 0 : defaultSocket_ = machinePara_.sockets[0];
302 0 : CHK_PTR_NULL(defaultSocket_);
303 :
304 0 : CHK_RET(hrtGetDeviceType(localDeviceType));
305 0 : HCCL_INFO("localDeviceType=[%d], remoteDeviceType=[%d]", localDeviceType, machinePara_.deviceType);
306 0 : CHK_RET(GetNicHandle());
307 :
308 : // 设置linkType
309 0 : transportAttr_.linkType = hccl::LinkType::LINK_ROCE;
310 :
311 : /* 获取当前的连接模式，offline模式或者op base模式 */
312 0 : workFlowMode_ = GetWorkflowMode();
313 0 : HCCL_INFO("current work mode is [%d]", workFlowMode_);
314 :
315 : /* 创建QP连接 */
316 0 : CHK_RET(InitQpConnect());
317 :
318 0 : HCCL_INFO("linkexp initialization success,Time:%lld us", DURATION_US(TIME_NOW() - startut));
319 :
320 0 : CHK_RET(GetQpAttr());
321 0 : return HCCL_SUCCESS;
322 : }
323 :
324 0 : HcclResult TransportDirectNpu::GetQpAttr()
325 : {
326 : char stackLogBuffer[LOG_TMPBUF_SIZE];
327 0 : s32 ret = snprintf_s(
328 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
329 : "communicator[%s], local rank[%u], ip[%s], remote rank[%u], ip[%s], transporttype[%s]",
330 : machinePara_.tag.c_str(), machinePara_.localUserrank, machinePara_.localIpAddr.GetReadableAddress(),
331 : machinePara_.remoteUserrank, machinePara_.remoteIpAddr.GetReadableAddress(),
332 0 : GetLinkTypeEnumStr(GetLinkType()).c_str());
333 0 : CHK_PRT_RET(
334 : ret == -1, HCCL_ERROR("[GetQpAttr]errNo[0x%016llx] sal snprintf_s error", HCCL_ERROR_CODE(HCCL_E_INTERNAL)),
335 : HCCL_E_INTERNAL);
336 0 : std::string logInfo = "create hccl transport:" + std::string(stackLogBuffer);
337 0 : for (u32 i = 0; i < qpHandles_.size(); i++) {
338 0 : struct QpAttr attr {};
339 0 : hrtRaGetQpAttr(qpHandles_[i], &attr);
340 0 : HCCL_USER_CRITICAL_LOG(
341 : "%s, rdma qpn[%u], rdma qp sport[%u], rdma TC[%u], rdma SL[%u]", logInfo.c_str(), attr.qpn, attr.udpSport,
342 : machinePara_.tc, machinePara_.sl);
343 : }
344 :
345 0 : return HCCL_SUCCESS;
346 0 : }
347 :
348 0 : HcclResult TransportDirectNpu::IsUseQpCreateWithAttrs(bool& isUseQpCreateWithAttrs, s32 qpMode)
349 : {
350 0 : isUseQpCreateWithAttrs = false;
351 0 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
352 0 : bool is910Bor91093
353 0 : = machinePara_.deviceType == DevType::DEV_TYPE_910B || machinePara_.deviceType == DevType::DEV_TYPE_910_93;
354 0 : if (is910Bor91093 && (qpMode == OFFLINE_QP_MODE_EXT || qpMode == OPBASE_QP_MODE_EXT)) {
355 0 : isUseQpCreateWithAttrs = true;
356 : }
357 : }
358 0 : return HCCL_SUCCESS;
359 : }
360 :
361 0 : HcclResult TransportDirectNpu::FillExchangeDataTotalSize()
362 : {
363 0 : const uint8_t memMsgCount = 3;
364 0 : exchangeDataTotalSize_ = 0;
365 0 : exchangeDataTotalSize_ += sizeof(u32); // 首个内容放qp数量
366 0 : exchangeDataTotalSize_ += sizeof(MemMsg) * memMsgCount; // output and input mem and aicpu
367 0 : exchangeDataTotalSize_ += machinePara_.exchangeInfo.size();
368 :
369 0 : HCCL_DEBUG("[TransportDirectNpu][FillExchangeDataTotalSize] exchangeDataTotalSize[%llu]", exchangeDataTotalSize_);
370 0 : return HCCL_SUCCESS;
371 : }
372 :
373 0 : HcclResult TransportDirectNpu::ConstructExchangeForSend()
374 : {
375 0 : exchangeDataForSend_.resize(exchangeDataTotalSize_);
376 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
377 0 : u64 exchangeDataBlankSize = exchangeDataTotalSize_;
378 : // 把qp对数量放在最前头，第一个做检验
379 0 : u32 qpNum = 1;
380 0 : s32 sRet = memcpy_s(exchangeDataPtr, sizeof(u32), reinterpret_cast<void*>(&qpNum), sizeof(u32));
381 0 : CHK_PRT_RET(
382 : sRet != EOK,
383 : HCCL_ERROR(
384 : "[Set][LocalMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
385 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
386 : HCCL_E_MEMORY);
387 0 : exchangeDataPtr += sizeof(u32);
388 0 : exchangeDataBlankSize -= sizeof(u32);
389 :
390 0 : CHK_RET(RegUserMem(MemType::AICPU_SYNC_MEM, exchangeDataPtr, exchangeDataBlankSize));
391 0 : CHK_RET(RegUserMem(MemType::USER_OUTPUT_MEM, exchangeDataPtr, exchangeDataBlankSize));
392 0 : CHK_RET(RegUserMem(MemType::USER_INPUT_MEM, exchangeDataPtr, exchangeDataBlankSize));
393 0 : CHK_RET(ConstructExchangeDataForSend(exchangeDataPtr, exchangeDataBlankSize));
394 :
395 0 : if (exchangeDataBlankSize != 0) {
396 0 : HCCL_ERROR(
397 : "[TransportDirectNpu][ConstructExchangeForSend] failed to construct exchange Data "
398 : "exchangeDataBlankSize[%llu]",
399 : exchangeDataBlankSize);
400 0 : return HCCL_E_INTERNAL;
401 : }
402 :
403 0 : HCCL_DEBUG("[TransportDirectNpu] ConstructExchangeForSend finished.");
404 0 : return HCCL_SUCCESS;
405 : }
406 :
407 0 : HcclResult TransportDirectNpu::ParseReceivedExchangeData()
408 : {
409 0 : u8* exchangeDataPtr = exchangeDataForRecv_.data();
410 0 : u64 exchangeDataBlankSize = exchangeDataTotalSize_;
411 :
412 : // 首先解析qp对数量，并作一致性校验
413 0 : u32 localQpNum = 1;
414 0 : u32 remoteQpNum = 0;
415 0 : s32 sRet = memcpy_s(reinterpret_cast<void*>(&remoteQpNum), sizeof(u32), exchangeDataPtr, sizeof(u32));
416 0 : CHK_PRT_RET(
417 : sRet != EOK,
418 : HCCL_ERROR(
419 : "[Get][RemoteMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
420 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
421 : HCCL_E_MEMORY);
422 0 : CHK_PRT_RET(
423 : localQpNum != remoteQpNum,
424 : HCCL_ERROR(
425 : "[TransportDirectNpu][ParseReceivedExchangeData]"
426 : "local qps[%u] not equal to remote qps[%u], rank:local[%u],remote[%u]",
427 : localQpNum, remoteQpNum, machinePara_.localUserrank, machinePara_.remoteUserrank),
428 : HCCL_E_INTERNAL);
429 0 : exchangeDataPtr += sizeof(u32);
430 0 : exchangeDataBlankSize -= sizeof(u32);
431 :
432 0 : CHK_RET(GetRemoteAddr(MemType::AICPU_SYNC_MEM, exchangeDataPtr, exchangeDataBlankSize));
433 :
434 0 : CHK_RET(GetRemoteAddr(MemType::USER_OUTPUT_MEM, exchangeDataPtr, exchangeDataBlankSize));
435 :
436 0 : CHK_RET(GetRemoteAddr(MemType::USER_INPUT_MEM, exchangeDataPtr, exchangeDataBlankSize));
437 :
438 0 : CHK_RET(ParseExchangeData(exchangeDataPtr, exchangeDataBlankSize));
439 :
440 0 : if (exchangeDataBlankSize != 0) {
441 0 : HCCL_ERROR(
442 : "[TransportDirectNpu][ParseReceivedExchangeData] failed to Parse exchange Data "
443 : "exchangeDataBlankSize[%llu]",
444 : exchangeDataBlankSize);
445 0 : return HCCL_E_INTERNAL;
446 : }
447 0 : HCCL_DEBUG("Parse Received ExchangeData success!");
448 0 : return HCCL_SUCCESS;
449 : }
450 :
451 0 : u32 TransportDirectNpu::GetQpsPerConnection()
452 : {
453 0 : u32 externalQps = std::max(static_cast<u32>(machinePara_.srcPorts.size()), 1U);
454 0 : s32 qpMode = GetQpMode();
455 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
456 0 : && externalQps != HCCL_QPS_PER_CONNECTION_DEFAULT) {
457 0 : HCCL_RUN_INFO(
458 : "HCCL_RDMA_QPS_PER_CONNECTION is set to [%u] but it is not effective in offline mode.", externalQps);
459 0 : } else if (qpMode != OPBASE_QP_MODE_EXT && externalQps > 1) {
460 0 : HCCL_RUN_INFO(
461 : "HCCL_RDMA_QPS_PER_CONNECTION is set to [%u] but current devType[%d] does not support multi-QP.",
462 : externalQps, machinePara_.deviceType);
463 0 : return 1; // 非单算子模式仅支持单QP， QPS = 1
464 : }
465 0 : if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
466 0 : return externalQps; // only work for opbase mode
467 : }
468 0 : return 1; // 非单算子模式仅支持单QP， QPS = 1
469 : }
470 :
471 0 : HcclResult TransportDirectNpu::GetNicHandle()
472 : {
473 0 : RaResourceInfo raResourceInfo;
474 0 : CHK_RET(NetworkManager::GetInstance(machinePara_.deviceLogicId).GetRaResourceInfo(raResourceInfo));
475 0 : std::map<HcclIpAddress, IpSocket>& tmpSocketMap = machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE ?
476 : raResourceInfo.nicSocketMap :
477 : raResourceInfo.hostNetSocketMap;
478 :
479 0 : HcclIpAddress localIpAddr = machinePara_.localIpAddr;
480 :
481 : // 获取 nicRdmaHandle
482 0 : auto itSocket = tmpSocketMap.find(localIpAddr);
483 0 : if (itSocket == tmpSocketMap.end()) {
484 0 : HCCL_ERROR(
485 : "[Get][NicHandle]In get nic handle, can not find socket handle, handle size[%u], "
486 : "local ip[%s]",
487 : tmpSocketMap.size(), localIpAddr.GetReadableAddress());
488 0 : return HCCL_E_PARA;
489 : }
490 :
491 0 : nicRdmaHandle_ = itSocket->second.nicRdmaHandle;
492 0 : CHK_PTR_NULL(nicRdmaHandle_);
493 :
494 0 : return HCCL_SUCCESS;
495 0 : }
496 :
497 : // 创建一个QP
498 0 : HcclResult TransportDirectNpu::CreateOneQp(
499 : s32 qpMode, u32 qpsPerConnection, QpHandle& qpHandle, AiQpInfo& aiQpInfo, bool useAicpu, u32 udpSport)
500 : {
501 0 : bool isUseQpCreateWithAttrs = false;
502 0 : CHK_RET(IsUseQpCreateWithAttrs(isUseQpCreateWithAttrs, qpMode));
503 : HcclResult ret;
504 0 : std::string useAicpuTitle = useAicpu ? std::string("aicpu ") : std::string("");
505 0 : std::string qpInfo = useAicpuTitle + std::string("rank:") + std::to_string(machinePara_.localWorldRank)
506 0 : + std::string(",localUserrank:") + std::to_string(machinePara_.localUserrank)
507 0 : + std::string(",localIpAddr: ") + std::string(machinePara_.localIpAddr.GetReadableAddress())
508 0 : + std::string(",deviceLogicId:") + std::to_string(machinePara_.deviceLogicId);
509 0 : struct QpExtAttrs attrs {};
510 : // 判断是否为NORMALQP需要使用qpMode_; hostnic场景的qpmode也是NORMALQP
511 0 : if (useAicpu || qpMode_ == QPMode::NORMAL) {
512 0 : bool isWorkFlowLib = (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB);
513 0 : CHK_RET(ConstructQpAttrs(qpMode, attrs, machinePara_.queueDepthAttr, isWorkFlowLib));
514 :
515 : // A3 aicpu图模式使用单个qp, qp深度为socket数量*128
516 0 : bool isAicpuLib = machinePara_.isAicpuModeEn && (machinePara_.deviceType == DevType::DEV_TYPE_910_93)
517 0 : && (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB);
518 0 : if (isAicpuLib) {
519 0 : attrs.qpAttr.cap.max_send_wr = machinePara_.sockets.size() * DEFAULT_OFFLINE_MAX_SEND_WR;
520 : }
521 0 : HCCL_DEBUG(
522 : "qp set max_send_wr %u, socket size %u, isWorkFlowLib %d", attrs.qpAttr.cap.max_send_wr,
523 : machinePara_.sockets.size(), isWorkFlowLib);
524 :
525 0 : attrs.udpSport = udpSport;
526 0 : ret = hrtRaAiQpCreate(machinePara_.localDeviceId, nicRdmaHandle_, &attrs, &aiQpInfo, qpHandle);
527 0 : HCCL_DEBUG("aiQpAddr:%llu db_index:%u, sq_index=%u", aiQpInfo.aiQpAddr, aiQpInfo.dbIndex, aiQpInfo.sqIndex);
528 0 : qpInfo = qpInfo + std::string(",sendCqDepth:") + std::to_string(attrs.cqAttr.sendCqDepth);
529 0 : } else if (!isUseQpCreateWithAttrs && qpsPerConnection == HCCL_QPS_PER_CONNECTION_DEFAULT) {
530 0 : ret = HrtRaQpCreate(nicRdmaHandle_, QP_FLAG_RC, qpMode, qpHandle);
531 0 : } else if (!isUseQpCreateWithAttrs && qpsPerConnection != HCCL_QPS_PER_CONNECTION_DEFAULT) {
532 0 : HCCL_ERROR("qpsPerConnection[%u] is set but qpMode[%d] is not supported", qpsPerConnection, qpMode);
533 0 : return HCCL_E_PARA;
534 : } else {
535 0 : CHK_RET(ConstructQpAttrs(qpMode, attrs, machinePara_.queueDepthAttr));
536 0 : attrs.udpSport = udpSport;
537 0 : ret = hrtRaQpCreateWithAttrs(nicRdmaHandle_, &attrs, qpHandle);
538 0 : qpInfo = qpInfo + std::string(",sendCqDepth:") + std::to_string(attrs.cqAttr.sendCqDepth);
539 : }
540 :
541 0 : RPT_ENV_ERR(
542 : ret != 0 || (qpHandle == nullptr), "EI0007", vector<string>({"resource_type", "resource_info"}),
543 : vector<string>({"qp", "CreateOneQp"}));
544 :
545 0 : CHK_PRT_RET(
546 : ret != HCCL_SUCCESS,
547 : HCCL_ERROR(
548 : "[%s][%s]create qp failed, localDeviceId[%d], qpMode[%d]", LOG_KEYWORDS_INIT_GROUP.c_str(),
549 : LOG_KEYWORDS_RESOURCE.c_str(), machinePara_.localDeviceId, qpMode),
550 : HCCL_E_ROCE_CONNECT);
551 :
552 : // 表示没有通过config配置，则使用环境变量配置
553 0 : CHK_RET(SetQpAttrQos(qpHandle, machinePara_.tc, machinePara_.sl));
554 : // 配置RDMA Timeout时间
555 0 : CHK_RET(SetQpAttrTimeOut(qpHandle));
556 : // 配置RDMA Retry Cnt重传次数
557 0 : CHK_RET(SetQpAttrRetryCnt(qpHandle));
558 : // qpn map 插入
559 0 : struct QpAttr attr {};
560 0 : CHK_RET(hrtRaGetQpAttr(qpHandle, &attr));
561 :
562 0 : g_qpn2IbversLinkMap_.Emplace(((static_cast<u64>(machinePara_.localDeviceId) << DEV_PHY_ID_BIT) | attr.qpn), this);
563 :
564 0 : HCCL_DEBUG("ra qp create success.");
565 0 : return HCCL_SUCCESS;
566 0 : }
567 :
568 0 : HcclResult TransportDirectNpu::CreateSingleQp(s32 qpMode) // 根据socket个数创建QP（下沉模板不够用多QP）
569 : {
570 0 : u32 socketNum = 1;
571 : // A3 aicpu图模式只使用1个qp，qp深度为socketNum*128，最大不超过32K
572 0 : bool isAicpuLib = machinePara_.isAicpuModeEn && (machinePara_.deviceType == DevType::DEV_TYPE_910_93)
573 0 : && (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB);
574 0 : if (!isAicpuLib) {
575 0 : socketNum = std::max(static_cast<u32>(machinePara_.sockets.size()), socketNum);
576 : }
577 : // 原来是 machinePara_.socketFdHandles 换成 machinePara_.sockets
578 0 : for (u32 i = 0; i < socketNum; i++) {
579 0 : QpHandle qpHandle = nullptr;
580 0 : u32 udpSport = machinePara_.srcPorts.empty() ? 0 : machinePara_.srcPorts[0];
581 0 : CHK_RET(CreateOneQp(
582 : qpMode, HCCL_QPS_PER_CONNECTION_DEFAULT, qpHandle, aiQpInfo_, machinePara_.isAicpuModeEn, udpSport));
583 0 : qpHandles_.push_back(qpHandle);
584 : }
585 0 : return HCCL_SUCCESS;
586 : }
587 :
588 0 : s32 TransportDirectNpu::GetQpMode()
589 : {
590 0 : s32 qpMode = NORMAL_QP_MODE;
591 :
592 0 : if (qpMode_ == QPMode::NORMAL) {
593 0 : return qpMode;
594 : }
595 :
596 0 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
597 0 : if (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
598 0 : qpMode = (machinePara_.deviceType == DevType::DEV_TYPE_910B
599 0 : || machinePara_.deviceType == DevType::DEV_TYPE_910_93) ?
600 : OPBASE_QP_MODE_EXT :
601 : OPBASE_QP_MODE;
602 : // isCapture需要创建下沉QP
603 0 : qpMode = (qpMode == OPBASE_QP_MODE_EXT && qpMode_ == QPMode::OFFLOAD) ? OFFLINE_QP_MODE_EXT : qpMode;
604 : } else {
605 0 : qpMode = (machinePara_.deviceType == DevType::DEV_TYPE_910B
606 0 : || machinePara_.deviceType == DevType::DEV_TYPE_910_93) ?
607 : OFFLINE_QP_MODE_EXT :
608 : OFFLINE_QP_MODE;
609 : }
610 : }
611 0 : if (machinePara_.isAicpuModeEn) {
612 0 : qpMode = (machinePara_.deviceType == DevType::DEV_TYPE_910B
613 0 : || machinePara_.deviceType == DevType::DEV_TYPE_910_93) ?
614 : OPBASE_QP_MODE_EXT :
615 : OPBASE_QP_MODE;
616 : }
617 0 : return qpMode;
618 : }
619 :
620 0 : HcclResult TransportDirectNpu::CreateQp()
621 : {
622 0 : s32 qpMode = GetQpMode();
623 0 : HCCL_DEBUG("[TransportDirectNpu][CreateQp] QpMode[%u]", qpMode);
624 0 : CHK_RET(CreateSingleQp(qpMode));
625 0 : HCCL_DEBUG("ra qp create %u qp success.", qpHandles_.size());
626 0 : return HCCL_SUCCESS;
627 : }
628 :
629 0 : HcclResult TransportDirectNpu::InitQpConnect()
630 : {
631 : /* 创建QP操作句柄 */
632 0 : qpsPerConnection_ = GetQpsPerConnection();
633 :
634 0 : CHK_RET(CreateQp());
635 :
636 0 : CHK_RET(FillExchangeDataTotalSize());
637 :
638 0 : CHK_RET(ConstructExchangeForSend());
639 :
640 0 : HCCL_DEBUG("[TransportDirectNpu] resource create done exchangeDataTotalSize_[%llu]", exchangeDataTotalSize_);
641 :
642 0 : HcclResult ret = defaultSocket_->Send(exchangeDataForSend_.data(), exchangeDataTotalSize_);
643 0 : CHK_PRT_RET(
644 : ret != HCCL_SUCCESS,
645 : HCCL_ERROR(
646 : "[TransportDirectNpu][InitQpConnect] failed to send exchangeData exchangeDataTotalSize[%llu], "
647 : "custom exchange data size [%llu].",
648 : exchangeDataTotalSize_, machinePara_.exchangeInfo.size()),
649 : ret);
650 0 : HCCL_DEBUG("[TransportDirectNpu]Seocket Send finished, exchangeDataTotalSize[%llu]", exchangeDataTotalSize_);
651 :
652 0 : exchangeDataForRecv_.resize(exchangeDataTotalSize_);
653 0 : ret = defaultSocket_->Recv(exchangeDataForRecv_.data(), exchangeDataTotalSize_);
654 0 : CHK_PRT_RET(
655 : ret != HCCL_SUCCESS,
656 : HCCL_ERROR(
657 : "[TransportDirectNpu][InitQpConnect] failed to recv exchangeData exchangeDataTotalSize[%llu], "
658 : "custom exchange data size [%llu].",
659 : exchangeDataTotalSize_, machinePara_.exchangeInfo.size()),
660 : ret);
661 :
662 0 : HCCL_DEBUG("[TransportDirectNpu][Init] Socket Data Recved");
663 :
664 0 : CHK_RET(ParseReceivedExchangeData());
665 :
666 : // 连接Qp
667 0 : CHK_RET(ConnectQp());
668 0 : HCCL_INFO("In link ibv, qp status has ready");
669 0 : return HCCL_SUCCESS;
670 : }
671 :
672 0 : HcclResult TransportDirectNpu::ConnectSingleQp(std::function<bool()> needStop)
673 : {
674 : // QP建链
675 0 : for (u32 i = 0; i < qpHandles_.size(); i++) {
676 0 : CHK_RET(HrtRaQpConnectAsync(qpHandles_[i], machinePara_.sockets[i]->GetFdHandle(), needStop));
677 : }
678 : // 查询QP建链是否成功
679 0 : s32 qpStatus = 0;
680 0 : s32 raRet = 0;
681 0 : auto startTime = std::chrono::steady_clock::now();
682 0 : HCCL_INFO("In link ibv, waiting for qp status ready...");
683 0 : for (u32 i = 0; i < qpHandles_.size(); i++) {
684 : while (true) {
685 0 : CHK_PRT_RET(needStop(), HCCL_ERROR("Terminating operation due to external request"), HCCL_E_INTERNAL);
686 :
687 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout_) {
688 0 : HCCL_ERROR("[Connect][Qp]get qp status timeout_=[%lld ms], qp_status=[%d]", timeout_, qpStatus);
689 0 : return HCCL_E_TIMEOUT;
690 : }
691 0 : raRet = hrtGetRaQpStatus(qpHandles_[i], &qpStatus);
692 0 : if ((!raRet) && (qpStatus == 1)) { // 为1时，qp 建链成功
693 0 : HCCL_INFO("In link ibv, %u of %u QP get status success.", (i + 1), qpHandles_.size());
694 0 : break;
695 : } else {
696 : // qp建链需要时间，获取qp状态直至超时
697 0 : SaluSleep(WAIT_US_COUNT);
698 : }
699 : }
700 : }
701 0 : return HCCL_SUCCESS;
702 : }
703 :
704 0 : HcclResult TransportDirectNpu::ConnectQp()
705 : {
706 0 : CHK_RET(ConnectSingleQp([this]() -> bool {
707 : return this->GetStopFlag();
708 : }));
709 0 : return HCCL_SUCCESS;
710 : }
711 :
712 0 : HcclResult TransportDirectNpu::TxAsync(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
713 : {
714 0 : CHK_RET(TxData(dstMemType, dstOffset, src, len, stream));
715 0 : return HCCL_SUCCESS;
716 : }
717 :
718 0 : HcclResult TransportDirectNpu::TxAsync(std::vector<TxMemoryInfo>& txMems, Stream& stream)
719 : {
720 0 : for (auto& mem : txMems) {
721 0 : CHK_RET(TxAsync(mem.dstMemType, mem.dstOffset, mem.src, mem.len, stream));
722 : }
723 0 : return HCCL_SUCCESS;
724 : }
725 :
726 0 : HcclResult TransportDirectNpu::RxAsync(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream)
727 : {
728 0 : CHK_PRT(RxData(srcMemType, srcOffset, dst, len, stream));
729 0 : return HCCL_SUCCESS;
730 : }
731 :
732 0 : HcclResult TransportDirectNpu::RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream& stream)
733 : {
734 0 : for (auto& mem : rxMems) {
735 0 : CHK_RET(RxAsync(mem.srcMemType, mem.srcOffset, mem.dst, mem.len, stream));
736 : }
737 0 : return HCCL_SUCCESS;
738 : }
739 :
740 0 : HcclResult TransportDirectNpu::DataReceivedAck(Stream& stream)
741 : {
742 0 : CHK_RET(PostFinAck(stream));
743 0 : CHK_RET(WaitFinAck(stream));
744 :
745 0 : return HCCL_SUCCESS;
746 : }
747 :
748 0 : HcclResult TransportDirectNpu::TxWaitDone(Stream& stream) { return HCCL_SUCCESS; }
749 :
750 : /* 发送ack消息(同步模式) */
751 0 : HcclResult TransportDirectNpu::TxAck(Stream& stream) { return HCCL_SUCCESS; }
752 :
753 : /* 接收ack消息(同步模式) */
754 0 : HcclResult TransportDirectNpu::RxAck(Stream& stream) { return HCCL_SUCCESS; }
755 :
756 0 : HcclResult TransportDirectNpu::TxDataSignal(Stream& stream) { return HCCL_SUCCESS; }
757 :
758 0 : HcclResult TransportDirectNpu::RxDataSignal(Stream& stream) { return HCCL_SUCCESS; }
759 :
760 0 : HcclResult TransportDirectNpu::RegUserMem(MemType memType, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
761 : {
762 0 : void* memPtr = nullptr;
763 : u64 memSize;
764 0 : switch (memType) {
765 0 : case MemType::USER_INPUT_MEM: {
766 0 : memPtr = machinePara_.inputMem.ptr();
767 0 : memSize = machinePara_.inputMem.size();
768 0 : break;
769 : }
770 :
771 0 : case MemType::USER_OUTPUT_MEM: {
772 0 : memPtr = machinePara_.outputMem.ptr();
773 0 : memSize = machinePara_.outputMem.size();
774 0 : break;
775 : }
776 :
777 0 : case MemType::AICPU_SYNC_MEM: {
778 0 : memPtr = aicpuMem_.ptr();
779 0 : memSize = aicpuMem_.size();
780 0 : break;
781 : }
782 :
783 0 : default: {
784 0 : HCCL_ERROR("[Reg][UserMem]not support dst_mem_type=%d", memType);
785 0 : return HCCL_E_NOT_SUPPORT;
786 : }
787 : }
788 0 : struct MrInfoT mrInfo = {nullptr};
789 0 : mrInfo.addr = memPtr;
790 0 : mrInfo.size = memSize;
791 0 : mrInfo.access = access_;
792 0 : for (u32 i = 0; i < qpHandles_.size(); i++) {
793 0 : CHK_RET(HrtRaMrReg(qpHandles_[i], &mrInfo));
794 : }
795 :
796 0 : memMsg_[static_cast<u32>(memType)].mrRegFlag = REG_VALID;
797 0 : memMsg_[static_cast<u32>(memType)].addr = memPtr;
798 0 : memMsg_[static_cast<u32>(memType)].len = memSize;
799 0 : memMsg_[static_cast<u32>(memType)].memType = memType;
800 0 : memMsg_[static_cast<u32>(memType)].lkey = mrInfo.lkey;
801 :
802 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
803 : exchangeDataPtr, exchangeDataBlankSize, reinterpret_cast<void*>(&memMsg_[static_cast<u32>(memType)]),
804 : sizeof(MemMsg)));
805 :
806 0 : exchangeDataPtr += sizeof(MemMsg);
807 0 : exchangeDataBlankSize -= sizeof(MemMsg);
808 :
809 0 : HCCL_DEBUG("memType=%d mem_ptr=%p mem_size=%llu Byte, key = %u", memType, memPtr, memSize, mrInfo.lkey);
810 :
811 0 : return HCCL_SUCCESS;
812 : }
813 :
814 0 : HcclResult TransportDirectNpu::GetMemInfo(UserMemType memType, void** dstMemPtr, u64* dstMemSize)
815 : {
816 0 : switch (memType) {
817 0 : case UserMemType::INPUT_MEM: {
818 0 : *dstMemPtr = remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].addr;
819 0 : *dstMemSize = remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].len;
820 0 : break;
821 : }
822 :
823 0 : case UserMemType::OUTPUT_MEM: {
824 0 : *dstMemPtr = remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].addr;
825 0 : *dstMemSize = remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].len;
826 0 : break;
827 : }
828 :
829 0 : default: {
830 0 : HCCL_ERROR("[Get][MemInfo]not support dst_mem_type=%d", memType);
831 0 : return HCCL_E_NOT_SUPPORT;
832 : }
833 : }
834 0 : return HCCL_SUCCESS;
835 : }
836 :
837 0 : HcclResult TransportDirectNpu::GetRemoteAddr(MemType memType, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
838 : {
839 0 : s32 sRet = memcpy_s(&remoteMemMsg_[static_cast<u32>(memType)], sizeof(MemMsg), exchangeDataPtr, sizeof(MemMsg));
840 0 : CHK_PRT_RET(
841 : sRet != EOK,
842 : HCCL_ERROR(
843 : "[Get][RemoteAddr]errNo[0x%016llx] In lbv exp get remote addr, "
844 : "memcpy failed. errorno[%d], params:destMaxSize[%zu],count[%zu]",
845 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(MemMsg), sizeof(MemMsg)),
846 : HCCL_E_MEMORY);
847 0 : CHK_PTR_NULL(remoteMemMsg_[static_cast<u32>(memType)].addr);
848 :
849 0 : exchangeDataPtr += sizeof(MemMsg);
850 0 : exchangeDataBlankSize -= sizeof(MemMsg);
851 0 : HCCL_INFO(
852 : "GetRemoteAddr success: memType=%d, addr=%p len=%llu, notifyId=%u", static_cast<int32_t>(memType),
853 : remoteMemMsg_[static_cast<u32>(memType)].addr, remoteMemMsg_[static_cast<u32>(memType)].len,
854 : remoteMemMsg_[static_cast<u32>(memType)].notifyId);
855 0 : return HCCL_SUCCESS;
856 : }
857 :
858 : /* 发送ack消息(同步模式) */
859 0 : HcclResult TransportDirectNpu::TxPrepare(Stream& stream) { return HCCL_SUCCESS; }
860 :
861 : /* 接收ack消息(同步模式) */
862 0 : HcclResult TransportDirectNpu::RxPrepare(Stream& stream) { return HCCL_SUCCESS; }
863 :
864 0 : HcclResult TransportDirectNpu::TxData(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
865 : {
866 0 : CHK_PTR_NULL(src);
867 : struct ApiParamDef {
868 : u32 lKey;
869 : u32 rKey;
870 : HcclQpInfoV2 qpInfo;
871 : u64 remoteAddr;
872 : u64 localAddr;
873 : u64 dataSize;
874 : u64 timeout;
875 : u64 localFlagAddr;
876 : u64 remoteFlagAddr;
877 : u32 lfKey;
878 : u32 rfKey;
879 : };
880 0 : const std::string kernelName = "RunTransportRoceTx";
881 0 : struct ApiParamDef apiParam = {};
882 0 : MemDetails inputMemDetails;
883 0 : MemDetails outputMemDetails;
884 0 : CHK_PRT(GetLocalMemDetails(UserMemType::INPUT_MEM, inputMemDetails));
885 0 : CHK_PRT(GetLocalMemDetails(UserMemType::OUTPUT_MEM, outputMemDetails));
886 :
887 0 : if (reinterpret_cast<u64>(src) >= inputMemDetails.addr
888 0 : && reinterpret_cast<u64>(src) < inputMemDetails.addr + inputMemDetails.size) {
889 0 : apiParam.lKey = inputMemDetails.key;
890 0 : } else if (
891 0 : reinterpret_cast<u64>(src) >= outputMemDetails.addr
892 0 : && reinterpret_cast<u64>(src) <= outputMemDetails.addr + outputMemDetails.size) {
893 0 : apiParam.lKey = outputMemDetails.key;
894 : } else {
895 0 : HCCL_ERROR(
896 : "[TransportDirectNpu][TxData]src_ptr=%p is out of range, inputmem src[%p], size[%llu];"
897 : " outputmem src[%p] size[%llu]",
898 : src, inputMemDetails.addr, inputMemDetails.size, outputMemDetails.addr, outputMemDetails.size);
899 0 : return HCCL_E_INTERNAL;
900 : }
901 0 : CHK_RET(GetRemoteMemKey(dstMemType, &apiParam.rKey));
902 0 : void* remoteAddr = nullptr;
903 0 : u64 memSize = 0;
904 0 : CHK_RET(GetMemInfo(dstMemType, &remoteAddr, &memSize));
905 0 : apiParam.remoteFlagAddr = reinterpret_cast<u64>(remoteMemMsg_[static_cast<u32>(MemType::AICPU_SYNC_MEM)].addr);
906 0 : apiParam.lfKey = memMsg_[static_cast<u32>(MemType::AICPU_SYNC_MEM)].lkey;
907 0 : apiParam.rfKey = remoteMemMsg_[static_cast<u32>(MemType::AICPU_SYNC_MEM)].lkey;
908 0 : apiParam.remoteAddr = reinterpret_cast<u64>(remoteAddr) + dstOffset;
909 0 : apiParam.dataSize = len;
910 0 : apiParam.localAddr = reinterpret_cast<u64>(src);
911 0 : apiParam.timeout = NOTIFY_DEFAULT_WAIT_TIME;
912 0 : if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET
913 0 : || dispatcher_->GetExecTimeOutSet()) {
914 0 : apiParam.timeout = dispatcher_->GetExecTimeOut();
915 : }
916 0 : apiParam.localFlagAddr = reinterpret_cast<u64>(aicpuMem_.ptr());
917 0 : std::vector<HcclQpInfoV2> aiQpInfos;
918 0 : CHK_RET(GetAiQpInfo(aiQpInfos));
919 0 : apiParam.qpInfo = aiQpInfos[0];
920 0 : HCCL_INFO(
921 : "[TransportDirectNpu][TxData]localRank %u remoteRank %u lkey %u rkey %u remoteAddr %p localAddr %p dataSize "
922 : "%llu "
923 : "timeout %llu localFlagAddr %p remoteFlagAddr %p lfkey %u rfkey %u qpinfo %llu",
924 : machinePara_.localUserrank, machinePara_.remoteUserrank, apiParam.lKey, apiParam.rKey, apiParam.remoteAddr,
925 : apiParam.localAddr, apiParam.dataSize, apiParam.timeout, apiParam.localFlagAddr, apiParam.remoteFlagAddr,
926 : apiParam.lfKey, apiParam.rfKey, apiParam.qpInfo.qpPtr);
927 :
928 : #ifndef CCL_KERNEL
929 : u16 timeOut = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
930 : std::numeric_limits<uint16_t>::max() :
931 : NOTIFY_DEFAULT_WAIT_TIME;
932 : CHK_PRT(AicpuAclKernelLaunch(
933 : stream.ptr(), reinterpret_cast<void*>(&apiParam), sizeof(apiParam), binHandle_, kernelName, true, timeOut));
934 : #else
935 0 : HCCL_ERROR("[AicpuAclKernelLaunch]Does not support this interface.");
936 0 : return HCCL_E_NOT_SUPPORT;
937 : #endif
938 : HCCL_INFO("[TransportDirectNpu][TxData] exec succ.");
939 : return HCCL_SUCCESS;
940 0 : }
941 :
942 0 : HcclResult TransportDirectNpu::RxData(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream)
943 : {
944 0 : CHK_PTR_NULL(dst);
945 : struct ApiParamDef {
946 : u32 lKey;
947 : u32 rKey;
948 : HcclQpInfoV2 qpInfo;
949 : u64 remoteAddr;
950 : u64 localAddr;
951 : u64 dataSize;
952 : u64 timeout;
953 : u64 localFlagAddr;
954 : u64 remoteFlagAddr;
955 : u32 lfKey;
956 : u32 rfKey;
957 : };
958 0 : const std::string kernelName = "RunTransportRoceRx";
959 0 : struct ApiParamDef apiParam = {};
960 0 : MemDetails inputMemDetails;
961 0 : MemDetails outputMemDetails;
962 0 : CHK_PRT(GetLocalMemDetails(UserMemType::INPUT_MEM, inputMemDetails));
963 0 : CHK_PRT(GetLocalMemDetails(UserMemType::OUTPUT_MEM, outputMemDetails));
964 :
965 0 : if (reinterpret_cast<u64>(dst) >= inputMemDetails.addr
966 0 : && reinterpret_cast<u64>(dst) < inputMemDetails.addr + inputMemDetails.size) {
967 0 : apiParam.lKey = inputMemDetails.key;
968 0 : } else if (
969 0 : reinterpret_cast<u64>(dst) >= outputMemDetails.addr
970 0 : && reinterpret_cast<u64>(dst) <= outputMemDetails.addr + outputMemDetails.size) {
971 0 : apiParam.lKey = outputMemDetails.key;
972 : } else {
973 0 : HCCL_ERROR(
974 : "[TransportDirectNpu][RxData]src_ptr=%p is out of range, inputmem src[%p], size[%llu];"
975 : " outputmem src[%p] size[%llu]",
976 : dst, inputMemDetails.addr, inputMemDetails.size, outputMemDetails.addr, outputMemDetails.size);
977 0 : return HCCL_E_INTERNAL;
978 : }
979 0 : CHK_RET(GetRemoteMemKey(srcMemType, &apiParam.rKey));
980 0 : void* remoteAddr = nullptr;
981 0 : u64 memSize = 0;
982 0 : CHK_RET(GetMemInfo(srcMemType, &remoteAddr, &memSize));
983 0 : apiParam.remoteFlagAddr = reinterpret_cast<u64>(remoteMemMsg_[static_cast<u32>(MemType::AICPU_SYNC_MEM)].addr);
984 0 : apiParam.lfKey = memMsg_[static_cast<u32>(MemType::AICPU_SYNC_MEM)].lkey;
985 0 : apiParam.rfKey = remoteMemMsg_[static_cast<u32>(MemType::AICPU_SYNC_MEM)].lkey;
986 0 : apiParam.remoteAddr = reinterpret_cast<u64>(remoteAddr) + srcOffset;
987 0 : apiParam.dataSize = len;
988 0 : apiParam.localAddr = reinterpret_cast<u64>(dst);
989 0 : apiParam.timeout = NOTIFY_DEFAULT_WAIT_TIME;
990 0 : if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET
991 0 : || dispatcher_->GetExecTimeOutSet()) {
992 0 : apiParam.timeout = dispatcher_->GetExecTimeOut();
993 : }
994 0 : apiParam.localFlagAddr = reinterpret_cast<u64>(aicpuMem_.ptr());
995 0 : std::vector<HcclQpInfoV2> aiQpInfos;
996 0 : CHK_RET(GetAiQpInfo(aiQpInfos));
997 0 : apiParam.qpInfo = aiQpInfos[0];
998 0 : HCCL_INFO(
999 : "[TransportDirectNpu][RxData]localRank %u remoteRank %u lkey %u rkey %u remoteAddr %p localAddr %p dataSize "
1000 : "%llu "
1001 : "timeout %llu localFlagAddr %p remoteFlagAddr %p lfkey %u rfkey %u qpinfo %llu",
1002 : machinePara_.localUserrank, machinePara_.remoteUserrank, apiParam.lKey, apiParam.rKey, apiParam.remoteAddr,
1003 : apiParam.localAddr, apiParam.dataSize, apiParam.timeout, apiParam.localFlagAddr, apiParam.remoteFlagAddr,
1004 : apiParam.lfKey, apiParam.rfKey, apiParam.qpInfo.qpPtr);
1005 :
1006 : #ifndef CCL_KERNEL
1007 : u16 timeOut = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
1008 : std::numeric_limits<uint16_t>::max() :
1009 : NOTIFY_DEFAULT_WAIT_TIME;
1010 : CHK_PRT(AicpuAclKernelLaunch(
1011 : stream.ptr(), reinterpret_cast<void*>(&apiParam), sizeof(apiParam), binHandle_, kernelName, true, timeOut));
1012 : #else
1013 0 : HCCL_ERROR("[AicpuAclKernelLaunch]Does not support this interface.");
1014 0 : return HCCL_E_NOT_SUPPORT;
1015 : #endif
1016 : return HCCL_SUCCESS;
1017 0 : }
1018 :
1019 0 : HcclResult TransportDirectNpu::TxDone(Stream& stream) { return HCCL_SUCCESS; }
1020 :
1021 0 : HcclResult TransportDirectNpu::RxDone(Stream& stream) { return HCCL_SUCCESS; }
1022 :
1023 0 : HcclResult TransportDirectNpu::PostFin(Stream& stream) { return HCCL_SUCCESS; }
1024 :
1025 0 : HcclResult TransportDirectNpu::WaitFin(Stream& stream) { return HCCL_SUCCESS; }
1026 :
1027 0 : HcclResult TransportDirectNpu::PostFinAck(Stream& stream) { return HCCL_SUCCESS; }
1028 :
1029 0 : HcclResult TransportDirectNpu::WaitFinAck(Stream& stream) { return HCCL_SUCCESS; }
1030 :
1031 0 : HcclResult TransportDirectNpu::GetRemoteMemKey(UserMemType memType, uint32_t* remoteMemKey)
1032 : {
1033 0 : switch (memType) {
1034 0 : case UserMemType::INPUT_MEM:
1035 : case UserMemType::OUTPUT_MEM:
1036 0 : *remoteMemKey = remoteMemMsg_[static_cast<u32>(memType)].lkey;
1037 0 : break;
1038 :
1039 0 : default:
1040 0 : HCCL_ERROR("[Get][RemoteMemKey]not support dst_mem_type=%d", memType);
1041 0 : return HCCL_E_NOT_SUPPORT;
1042 : }
1043 0 : return HCCL_SUCCESS;
1044 : }
1045 :
1046 0 : HcclResult TransportDirectNpu::GetLocalMemDetails(UserMemType memType, MemDetails& memDetails)
1047 : {
1048 0 : switch (memType) {
1049 0 : case UserMemType::INPUT_MEM:
1050 : case UserMemType::OUTPUT_MEM:
1051 0 : memDetails.addr = reinterpret_cast<u64>(memMsg_[static_cast<u32>(memType)].addr);
1052 0 : memDetails.size = memMsg_[static_cast<u32>(memType)].len;
1053 0 : memDetails.key = memMsg_[static_cast<u32>(memType)].lkey;
1054 0 : break;
1055 :
1056 0 : default:
1057 0 : HCCL_ERROR("[Get][LocalMemDetails]not support dst_mem_type=%d", memType);
1058 0 : return HCCL_E_NOT_SUPPORT;
1059 : }
1060 0 : return HCCL_SUCCESS;
1061 : }
1062 :
1063 0 : HcclResult TransportDirectNpu::GetAiQpInfo(std::vector<HcclQpInfoV2>& aiQpInfo)
1064 : {
1065 0 : aiQpInfo.resize(aiQpInfos_.size() + 1);
1066 :
1067 0 : aiQpInfo[0].qpPtr = aiQpInfo_.aiQpAddr;
1068 0 : aiQpInfo[0].sqIndex = aiQpInfo_.sqIndex;
1069 0 : aiQpInfo[0].dbIndex = aiQpInfo_.dbIndex;
1070 0 : HCCL_DEBUG(
1071 : "[TransportDirectNpu][GetAiQpInfo] i[0] qpPtr[%llu] sqIndex[%u] dbIndex[%u]", aiQpInfo[0].qpPtr,
1072 : aiQpInfo[0].sqIndex, aiQpInfo[0].dbIndex);
1073 0 : for (u32 i = 1, j = 0; i < aiQpInfo.size(); i++, j++) {
1074 0 : aiQpInfo[i].qpPtr = aiQpInfos_[j].aiQpAddr;
1075 0 : aiQpInfo[i].sqIndex = aiQpInfos_[j].sqIndex;
1076 0 : aiQpInfo[i].dbIndex = aiQpInfos_[j].dbIndex;
1077 0 : HCCL_DEBUG(
1078 : "[TransportDirectNpu][GetAiQpInfo] i[%u] qpPtr[%llu] sqIndex[%u] dbIndex[%u]", i, aiQpInfo[i].qpPtr,
1079 : aiQpInfo[i].sqIndex, aiQpInfo[i].dbIndex);
1080 : }
1081 0 : return HCCL_SUCCESS;
1082 : }
1083 :
1084 0 : HcclResult TransportDirectNpu::GetTransportErrorCqe(
1085 : const HcclNetDevCtx netDevCtx, std::vector<std::pair<TransportBase*, CqeInfo>>& infos, u32& num)
1086 : {
1087 0 : if (g_qpn2IbversLinkMap_.Size() == 0) {
1088 0 : num = 0;
1089 0 : return HCCL_SUCCESS;
1090 : }
1091 :
1092 0 : if (UNLIKELY(!g_flag)) {
1093 0 : CHK_RET(IsSuppCqeErrInfoListConfig(g_isSupCqeErrInfoListConfig));
1094 0 : g_flag = true;
1095 : }
1096 :
1097 0 : CHK_PTR_NULL(netDevCtx);
1098 0 : s32 deviceLogicId = (static_cast<NetDevContext*>(netDevCtx))->GetLogicId();
1099 0 : s32 devicePhyId = (static_cast<NetDevContext*>(netDevCtx))->GetPhyId();
1100 0 : HcclIpAddress localIp = (static_cast<NetDevContext*>(netDevCtx))->GetLocalIp();
1101 0 : NicType nicType = (static_cast<NetDevContext*>(netDevCtx))->GetNicType();
1102 0 : CHK_PRT_RET(
1103 : nicType == NicType::HOST_NIC_TYPE,
1104 : HCCL_WARNING("[TransportDirectNpu][GetTransportErrorCqe] nicType[%d] not support", nicType), HCCL_SUCCESS);
1105 0 : RaResourceInfo raResourceInfo;
1106 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId).GetRaResourceInfo(raResourceInfo));
1107 0 : RdmaHandle rdmaHandle = raResourceInfo.nicSocketMap[localIp].nicRdmaHandle;
1108 0 : CHK_PTR_NULL(rdmaHandle);
1109 :
1110 0 : if (g_isSupCqeErrInfoListConfig) {
1111 0 : u32 loop = 0;
1112 0 : if (num > CQE_ARRAY_SIZE) {
1113 0 : loop = (num % CQE_ARRAY_SIZE) ? (num / CQE_ARRAY_SIZE) : ((num / CQE_ARRAY_SIZE) - 1);
1114 : }
1115 :
1116 0 : struct CqeErrInfo infolist[CQE_ARRAY_SIZE] = {};
1117 0 : u32 cqeNum = CQE_ARRAY_SIZE;
1118 0 : for (u32 index = 0; index <= loop; index++) {
1119 0 : cqeNum = (index == loop) ? (num - index * CQE_ARRAY_SIZE) : CQE_ARRAY_SIZE;
1120 0 : u32 temNum = cqeNum;
1121 0 : CHK_RET(hrtRaGetCqeErrInfoList(rdmaHandle, infolist, &temNum));
1122 0 : ProcessCqeInfo(devicePhyId, infolist, temNum, infos);
1123 0 : if (temNum < cqeNum) {
1124 0 : break;
1125 : }
1126 : }
1127 : } else {
1128 0 : struct CqeErrInfo infolist[1] = {};
1129 0 : CHK_RET(hrtRaGetCqeErrInfo(devicePhyId, &infolist[0]));
1130 0 : if (infolist[0].status == 0) {
1131 0 : num = 0;
1132 0 : return HCCL_SUCCESS;
1133 : }
1134 0 : u32 cqeNum = 1;
1135 0 : ProcessCqeInfo(devicePhyId, infolist, cqeNum, infos);
1136 : }
1137 :
1138 0 : num = infos.size();
1139 :
1140 0 : return HCCL_SUCCESS;
1141 0 : }
1142 :
1143 0 : void TransportDirectNpu::ProcessCqeInfo(
1144 : const s32 deviceId, const struct CqeErrInfo* infolist, const u32 cqeNum,
1145 : std::vector<std::pair<TransportBase*, CqeInfo>>& infos)
1146 : {
1147 0 : for (u32 i = 0; i < cqeNum; i++) {
1148 : // localPhyId + qpn
1149 0 : auto it = g_qpn2IbversLinkMap_.Find(((static_cast<u64>(deviceId) << DEV_PHY_ID_BIT) | infolist[i].qpn));
1150 0 : if (it.second) {
1151 0 : TransportBase* ptr = reinterpret_cast<TransportBase*>(it.first->second);
1152 0 : infos.push_back(
1153 0 : std::make_pair(ptr, CqeInfo(infolist[i].time, infolist[i].status, it.first->second->GetRemoteIp())));
1154 0 : cqeErrQpn_ = infolist[i].qpn;
1155 : } else {
1156 0 : HCCL_RUN_WARNING("[GetTransportErrorCqe]get err failed, transport is not find.");
1157 : }
1158 : }
1159 0 : return;
1160 : }
1161 :
1162 0 : HcclIpAddress& TransportDirectNpu::GetRemoteIp() { return machinePara_.remoteIpAddr; }
1163 :
1164 0 : HcclResult TransportDirectNpu::GetTransportId(u32& id)
1165 : {
1166 0 : id = cqeErrQpn_;
1167 0 : return HCCL_SUCCESS;
1168 : }
1169 : } // namespace hccl
|
