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