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