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 "device_capacity.h"
16 : #include "network_manager_pub.h"
17 : #include "adapter_rts.h"
18 : #include "externalinput_pub.h"
19 : #include "hccl_network.h"
20 : #include "externalinput.h"
21 : #include "../host/transport_ibverbs.h"
22 :
23 : // 混合模式（RoCE Cross-Mode）公共类型定义
24 : #include "../../../../base_comm/resources/endpoint_pairs/channels/host/exchange_data_format.h"
25 :
26 : using namespace std;
27 : constexpr u32 RDMA_QP_EXPECT_STATUS_PAUSE = 5;
28 : constexpr u32 RDMA_QP_EXPECT_STATUS_CONNECTED = 1;
29 :
30 : namespace hccl {
31 : std::array<DeviceMem, MAX_MODULE_DEVICE_NUM> TransportIbverbs::notifyValueMem_;
32 : std::array<std::mutex, MAX_MODULE_DEVICE_NUM> TransportIbverbs::notifyValueMutex_;
33 : std::array<Referenced, MAX_MODULE_DEVICE_NUM> TransportIbverbs::instanceRef_;
34 : UniversalConcurrentMap<u64, TransportIbverbs*> TransportIbverbs::g_qpn2IbversLinkMap_;
35 : bool TransportIbverbs::g_flag = false;
36 : bool TransportIbverbs::g_isSupCqeErrInfoListConfig = false;
37 : u32 TransportIbverbs::cqeErrQpn_ = 0;
38 :
39 : constexpr u32 CQE_ARRAY_SIZE = 128;
40 : constexpr u32 DEV_PHY_ID_BIT = 32;
41 :
42 : constexpr u32 WQE_RESERVE_LENGTH = 4;
43 :
44 : constexpr u32 NOTIFY_VA_ALIGN_EIGHT = 8; // notifyVa地址8byte对齐
45 :
46 23 : TransportIbverbs::TransportIbverbs(
47 : DispatcherPub* dispatcher, const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara,
48 23 : std::chrono::milliseconds timeout)
49 : : TransportNet(dispatcher, notifyPool, machinePara, timeout),
50 23 : qpsPerConnection_(1),
51 23 : notifySize_(0),
52 23 : ackNotify_(nullptr),
53 23 : dataAckNotify_(nullptr),
54 23 : access_(RA_ACCESS_LOCAL_WRITE | RA_ACCESS_REMOTE_WRITE | RA_ACCESS_REMOTE_READ),
55 23 : workFlowMode_(HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB),
56 23 : sqeCounter_(0),
57 23 : currentQP_(0),
58 46 : qpMode_(machinePara.qpMode)
59 : {
60 23 : dataNotify_ = nullptr;
61 23 : if (machinePara_.deviceLogicId >= 0 && (static_cast<u32>(machinePara_.deviceLogicId) < MAX_MODULE_DEVICE_NUM)) {
62 23 : instanceRef_[machinePara_.deviceLogicId].Ref();
63 : }
64 23 : }
65 :
66 23 : TransportIbverbs::~TransportIbverbs()
67 : {
68 23 : HCCL_DEBUG("~TransportIbverbs Enter!");
69 :
70 23 : (void)DeInit();
71 :
72 23 : if (machinePara_.deviceLogicId >= 0 && (static_cast<u32>(machinePara_.deviceLogicId) < MAX_MODULE_DEVICE_NUM)) {
73 23 : if (instanceRef_[machinePara_.deviceLogicId].Unref() == 0) {
74 3 : std::unique_lock<std::mutex> lock(notifyValueMutex_[machinePara_.deviceLogicId]);
75 3 : notifyValueMem_[machinePara_.deviceLogicId].free();
76 3 : }
77 : }
78 23 : HCCL_DEBUG("~TransportIbverbs Success!");
79 23 : }
80 :
81 34 : HcclResult TransportIbverbs::DeInit()
82 : {
83 34 : (void)DeRegMR();
84 :
85 34 : (void)DestroySignal();
86 :
87 34 : (void)DestroyQP();
88 :
89 34 : return HCCL_SUCCESS;
90 : }
91 :
92 0 : HcclResult TransportIbverbs::DeRegOneMR(QpHandle& qpHandle, MemMsg& memMsg)
93 : {
94 0 : struct MrInfoT mrInfo = {};
95 0 : mrInfo.addr = memMsg.addr;
96 0 : HcclResult ret = HrtRaMrDereg(qpHandle, &mrInfo);
97 0 : CHK_PRT_RET(
98 : ret != HCCL_SUCCESS,
99 : HCCL_ERROR("errNo[0x%016llx] in link lbv, In lbv exp deconstruct, mr dereg failed.", HCCL_ERROR_CODE(ret)),
100 : ret);
101 0 : return HCCL_SUCCESS;
102 : }
103 :
104 1 : void TransportIbverbs::DeRegMRForQPhandles(MemMsg& memMsg)
105 : {
106 1 : for (u32 j = 0; j < combineQpHandles_.size(); j++) {
107 0 : if (combineQpHandles_[j].qpHandle == nullptr) {
108 0 : continue;
109 : }
110 0 : (void)DeRegOneMR(combineQpHandles_[j].qpHandle, memMsg);
111 : }
112 1 : for (u32 j = 0; j < multiCombineQpHandles_.size(); j++) {
113 0 : if (multiCombineQpHandles_[j].qpHandle == nullptr) {
114 0 : continue;
115 : }
116 0 : (void)DeRegOneMR(multiCombineQpHandles_[j].qpHandle, memMsg);
117 : }
118 1 : }
119 :
120 34 : HcclResult TransportIbverbs::DeRegMR()
121 : {
122 : /* 销毁mr */
123 34 : std::map<uintptr_t, s32> addrMap;
124 578 : for (s32 i = 0; i < static_cast<s32>(MemType::MEM_TYPE_RESERVED); i++) {
125 544 : if (memMsg_[i].mrRegFlag == REG_VALID) {
126 : std::pair<std::map<uintptr_t, s32>::iterator, bool> res
127 1 : = addrMap.insert(std::pair<uintptr_t, s32>(reinterpret_cast<uintptr_t>(memMsg_[i].addr), 0));
128 1 : if (res.second) {
129 1 : DeRegMRForQPhandles(memMsg_[i]);
130 : }
131 : }
132 : }
133 34 : return HCCL_SUCCESS;
134 34 : }
135 :
136 0 : HcclResult TransportIbverbs::DestroyQP(QpHandle& qpHandle)
137 : {
138 0 : if (qpHandle != nullptr) {
139 0 : struct QpAttr attr {};
140 0 : CHK_RET(hrtRaGetQpAttr(qpHandle, &attr));
141 :
142 0 : g_qpn2IbversLinkMap_.Erase(((static_cast<u64>(machinePara_.localDeviceId) << DEV_PHY_ID_BIT) | attr.qpn));
143 :
144 0 : HcclResult ret = HrtRaQpDestroy(qpHandle);
145 0 : if (ret != HCCL_SUCCESS) {
146 0 : HCCL_ERROR("errNo[0x%016llx] in link lbv, lbv exp deconstruct, qp destroy failed.", HCCL_ERROR_CODE(ret));
147 : }
148 0 : qpHandle = nullptr;
149 : }
150 0 : return HCCL_SUCCESS;
151 : }
152 :
153 34 : HcclResult TransportIbverbs::DestroyQP()
154 : {
155 34 : for (u32 i = 0; i < combineQpHandles_.size(); i++) {
156 0 : CHK_RET(DestroyQP(combineQpHandles_[i].qpHandle));
157 : }
158 34 : for (u32 i = 0; i < multiCombineQpHandles_.size(); i++) {
159 0 : CHK_RET(DestroyQP(multiCombineQpHandles_[i].qpHandle));
160 : }
161 34 : return HCCL_SUCCESS;
162 : }
163 :
164 1 : HcclResult TransportIbverbs::Stop()
165 : {
166 1 : HcclResult ret = hrtRaQpBatchModify(
167 1 : nicRdmaHandle_, &combineQpHandles_[0].qpHandle, combineQpHandles_.size(), RDMA_QP_EXPECT_STATUS_PAUSE);
168 1 : if (ret != HCCL_SUCCESS) {
169 0 : HCCL_ERROR("errNo[0x%016llx] in link lbv, ra qp modify stop fail.", HCCL_ERROR_CODE(ret));
170 0 : return HCCL_E_INTERNAL;
171 : }
172 1 : return HCCL_SUCCESS;
173 : }
174 :
175 1 : HcclResult TransportIbverbs::Resume()
176 : {
177 1 : HcclResult ret = hrtRaQpBatchModify(
178 1 : nicRdmaHandle_, &combineQpHandles_[0].qpHandle, combineQpHandles_.size(), RDMA_QP_EXPECT_STATUS_CONNECTED);
179 1 : if (ret != HCCL_SUCCESS) {
180 0 : HCCL_ERROR("errNo[0x%016llx] in link lbv, ra qp modify resume fail.", HCCL_ERROR_CODE(ret));
181 0 : return HCCL_E_INTERNAL;
182 : }
183 1 : return HCCL_SUCCESS;
184 : }
185 :
186 1 : HcclResult TransportIbverbs::Init()
187 : {
188 1 : HCCL_INFO(
189 : "machineType=[%d], serverId=[%s], localDeviceId=[%d], remoteDeviceId=[%d], "
190 : "localRank=[%u], localUserRank=[%u], remoteRank=[%u], remoteUserrank=[%u], "
191 : "deviceType=[%d], inputMem=[%p], outputMem=[%p], isAicpuModeEn[%d], notifyNum[%u], "
192 : "isIndOp[%d], custom exchange data size [%llu], drainEnable[%d]",
193 : machinePara_.machineType, machinePara_.serverId.c_str(), machinePara_.localDeviceId,
194 : machinePara_.remoteDeviceId, machinePara_.localUserrank, machinePara_.localWorldRank,
195 : machinePara_.remoteUserrank, machinePara_.remoteWorldRank, machinePara_.deviceType, machinePara_.inputMem.ptr(),
196 : machinePara_.outputMem.ptr(), machinePara_.isAicpuModeEn, machinePara_.notifyNum, machinePara_.isIndOp,
197 : machinePara_.exchangeInfo.size(), machinePara_.drainEnable);
198 1 : HcclUs startut = TIME_NOW();
199 :
200 1 : if (machinePara_.userMemEnable) {
201 1 : CHK_SMART_PTR_NULL(machinePara_.inputMem);
202 0 : CHK_SMART_PTR_NULL(machinePara_.outputMem);
203 0 : CHK_SMART_PTR_NULL(notifyPool_);
204 : }
205 :
206 0 : if (machinePara_.drainEnable) {
207 0 : CHK_SMART_PTR_NULL(notifyPool_);
208 : }
209 :
210 0 : CHK_PTR_NULL(dispatcher_);
211 0 : CHK_RET(CheckDeviceId());
212 0 : CHK_RET(CheckExchangeData());
213 :
214 : // 上层初始化时保证 machinePara_.sockets 非空
215 0 : if (machinePara_.sockets.size() == 0) {
216 0 : HCCL_ERROR("machinePara sockets is empty.");
217 0 : return HCCL_E_INTERNAL;
218 : }
219 0 : defaultSocket_ = machinePara_.sockets[0];
220 0 : CHK_PTR_NULL(defaultSocket_);
221 :
222 0 : CHK_RET(hrtGetDeviceType(localDeviceType));
223 0 : HCCL_INFO("localDeviceType=[%d], remoteDeviceType=[%d]", localDeviceType, machinePara_.deviceType);
224 0 : CHK_RET(GetNicHandle());
225 :
226 : // 设置linkType
227 0 : transportAttr_.linkType = hccl::LinkType::LINK_ROCE;
228 :
229 : /* 获取当前的连接模式，offline模式或者op base模式 */
230 0 : workFlowMode_ = GetWorkflowMode();
231 0 : HCCL_INFO("current work mode is [%d]", workFlowMode_);
232 :
233 0 : CHK_RET(GetNotifySize());
234 :
235 : /* 创建QP连接 */
236 0 : CHK_RET(InitQpConnect());
237 :
238 0 : HCCL_INFO("linkexp initialization success,Time:%lld us", DURATION_US(TIME_NOW() - startut));
239 :
240 0 : CHK_RET(GetQpAttr());
241 0 : return HCCL_SUCCESS;
242 : }
243 :
244 0 : HcclResult TransportIbverbs::GetQpAttr()
245 : {
246 : char stackLogBuffer[LOG_TMPBUF_SIZE];
247 0 : s32 ret = snprintf_s(
248 : stackLogBuffer, LOG_TMPBUF_SIZE, LOG_TMPBUF_SIZE - 1U,
249 : "communicator[%s], local rank[%u], ip[%s], remote rank[%u], ip[%s], transporttype[%s]",
250 : machinePara_.tag.c_str(), machinePara_.localUserrank, machinePara_.localIpAddr.GetReadableAddress(),
251 : machinePara_.remoteUserrank, machinePara_.remoteIpAddr.GetReadableAddress(),
252 0 : GetLinkTypeEnumStr(GetLinkType()).c_str());
253 0 : CHK_PRT_RET(
254 : ret == -1, HCCL_ERROR("[GetQpAttr]errNo[0x%016llx] sal snprintf_s error", HCCL_ERROR_CODE(HCCL_E_INTERNAL)),
255 : HCCL_E_INTERNAL);
256 0 : std::string logInfo = "create hccl transport:" + std::string(stackLogBuffer);
257 0 : for (u32 i = 0; i < combineQpHandles_.size(); i++) {
258 0 : struct QpAttr attr {};
259 0 : CHK_RET(hrtRaGetQpAttr(combineQpHandles_[i].qpHandle, &attr));
260 0 : HCCL_USER_CRITICAL_LOG(
261 : "%s, rdma qpn[%u], rdma qp sport[%u], rdma TC[%u], rdma SL[%u]", logInfo.c_str(), attr.qpn, attr.udpSport,
262 : machinePara_.tc, machinePara_.sl);
263 : }
264 0 : if (UseMultiQp()) {
265 0 : for (u32 i = 0; i < multiCombineQpHandles_.size(); i++) {
266 0 : struct QpAttr attr {};
267 0 : CHK_RET(hrtRaGetQpAttr(multiCombineQpHandles_[i].qpHandle, &attr));
268 0 : HCCL_USER_CRITICAL_LOG(
269 : "%s, rdma qpn[%u], rdma qp sport[%u], rdma TC[%u], rdma SL[%u]", logInfo.c_str(), attr.qpn,
270 : attr.udpSport, machinePara_.tc, machinePara_.sl);
271 : }
272 : }
273 0 : return HCCL_SUCCESS;
274 0 : }
275 :
276 0 : HcclResult TransportIbverbs::GetNotifySize()
277 : {
278 0 : u32 notifySize = 0;
279 0 : CHK_RET(hrtGetNotifySize(notifySize));
280 0 : notifySize_ = notifySize;
281 0 : return HCCL_SUCCESS;
282 : }
283 :
284 0 : HcclResult TransportIbverbs::IsUseQpCreateWithAttrs(bool& isUseQpCreateWithAttrs, s32 qpMode)
285 : {
286 0 : isUseQpCreateWithAttrs = false;
287 0 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
288 0 : bool is910Bor91093
289 0 : = machinePara_.deviceType == DevType::DEV_TYPE_910B || machinePara_.deviceType == DevType::DEV_TYPE_910_93;
290 0 : if (is910Bor91093 && (qpMode == OFFLINE_QP_MODE_EXT || qpMode == OPBASE_QP_MODE_EXT)) {
291 0 : isUseQpCreateWithAttrs = true;
292 : }
293 : }
294 0 : return HCCL_SUCCESS;
295 : }
296 :
297 8 : HcclResult TransportIbverbs::FillExchangeDataTotalSize()
298 : {
299 8 : exchangeDataTotalSize_ = 0;
300 8 : exchangeDataTotalSize_ += sizeof(u32); // 首个内容放qp数量
301 8 : if (UseMultiQp()) {
302 2 : exchangeDataTotalSize_ += sizeof(u32); // 再放个MultiQpThreshold
303 : }
304 8 : if (machinePara_.userMemEnable) {
305 4 : exchangeDataTotalSize_ += sizeof(MemMsg) * 2; // 2: output and input mem
306 4 : exchangeDataTotalSize_ += sizeof(MemMsg) * 3; // 3: dataNotify_\ackNotify_\dataAckNotify_
307 4 : if (UseMultiQp()) {
308 1 : exchangeDataTotalSize_ += qpsPerConnection_ * sizeof(MemMsg); // 多QP下新增协商内容
309 : }
310 : }
311 8 : if (machinePara_.drainEnable) {
312 1 : exchangeDataTotalSize_ += sizeof(MemMsg); // Drain用到了dataNotify_
313 1 : exchangeDataTotalSize_ += sizeof(MemMsg); // 本端notify srcmem, 用于Drain
314 : }
315 8 : if (!isHybridMode_) {
316 8 : exchangeDataTotalSize_ += machinePara_.exchangeInfo.size();
317 : }
318 :
319 : // 4.新增notify资源统计
320 : // 单qp notify资源大小：1*notifyNum， 多qp notify资源大小：qpNum*notifyNum
321 8 : exchangeDataTotalSize_ += qpsPerConnection_ * sizeof(MemMsg) * notifyNum_;
322 :
323 : // 5.新增和对端协商atomic write是否使能
324 8 : exchangeDataTotalSize_ += sizeof(u8);
325 :
326 8 : if (machinePara_.isIndOp) {
327 : // 6. userDeviceMem数量\userDeviceMem\userHostMem数量\userHostMem
328 0 : exchangeDataTotalSize_ += sizeof(u32);
329 0 : exchangeDataTotalSize_ += sizeof(MemMsg) * machinePara_.userDeviceMem.size();
330 0 : exchangeDataTotalSize_ += sizeof(u32);
331 0 : exchangeDataTotalSize_ += sizeof(MemMsg) * machinePara_.userHostMem.size();
332 : }
333 :
334 8 : HCCL_DEBUG("[TransportIbverbs][FillExchangeDataTotalSize] exchangeDataTotalSize[%llu]", exchangeDataTotalSize_);
335 8 : return HCCL_SUCCESS;
336 : }
337 :
338 0 : HcclResult TransportIbverbs::ConstructExchangeForSend()
339 : {
340 0 : exchangeDataForSend_.resize(exchangeDataTotalSize_);
341 0 : u8* exchangeDataPtr = exchangeDataForSend_.data();
342 0 : u64 exchangeDataBlankSize = exchangeDataTotalSize_;
343 : // 把qp对数量放在最前头，第一个做检验
344 0 : u32 qpNum = UseMultiQp() ? qpsPerConnection_ : 1;
345 0 : s32 sRet = memcpy_s(exchangeDataPtr, sizeof(u32), reinterpret_cast<void*>(&qpNum), sizeof(u32));
346 0 : CHK_PRT_RET(
347 : sRet != EOK,
348 : HCCL_ERROR(
349 : "[Set][LocalMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
350 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
351 : HCCL_E_MEMORY);
352 0 : exchangeDataPtr += sizeof(u32);
353 0 : exchangeDataBlankSize -= sizeof(u32);
354 :
355 0 : if (UseMultiQp()) {
356 : // 把multiQpThreshold放第二个，第二个做检验
357 0 : u32 multiQpThreshold = GetExternalInputMultiQpThreshold();
358 0 : sRet = memcpy_s(exchangeDataPtr, sizeof(u32), reinterpret_cast<void*>(&multiQpThreshold), sizeof(u32));
359 0 : CHK_PRT_RET(
360 : sRet != EOK,
361 : HCCL_ERROR(
362 : "[Set][LocalMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
363 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
364 : HCCL_E_MEMORY);
365 0 : exchangeDataPtr += sizeof(u32);
366 0 : exchangeDataBlankSize -= sizeof(u32);
367 : }
368 :
369 0 : if (machinePara_.userMemEnable) {
370 0 : CHK_RET(RegUserMem(MemType::USER_OUTPUT_MEM, exchangeDataPtr, exchangeDataBlankSize));
371 0 : CHK_RET(RegUserMem(MemType::USER_INPUT_MEM, exchangeDataPtr, exchangeDataBlankSize));
372 0 : if (machinePara_.isAicpuModeEn) {
373 0 : CHK_RET(CreateNotifyBuffer(
374 : dataNotify_, MemType::DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize,
375 : NotifyLoadType::DEVICE_NOTIFY));
376 0 : CHK_RET(CreateNotifyBuffer(
377 : ackNotify_, MemType::ACK_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize,
378 : NotifyLoadType::DEVICE_NOTIFY));
379 0 : CHK_RET(CreateNotifyBuffer(
380 : dataAckNotify_, MemType::DATA_ACK_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize,
381 : NotifyLoadType::DEVICE_NOTIFY));
382 : } else {
383 0 : CHK_RET(CreateNotifyBuffer(dataNotify_, MemType::DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
384 0 : CHK_RET(CreateNotifyBuffer(ackNotify_, MemType::ACK_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
385 0 : CHK_RET(CreateNotifyBuffer(
386 : dataAckNotify_, MemType::DATA_ACK_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
387 : }
388 : }
389 0 : if (!machinePara_.userMemEnable && machinePara_.drainEnable) {
390 0 : if (machinePara_.isAicpuModeEn) {
391 0 : CHK_RET(CreateNotifyBuffer(
392 : dataNotify_, MemType::DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize,
393 : NotifyLoadType::DEVICE_NOTIFY));
394 : } else {
395 0 : CHK_RET(CreateNotifyBuffer(dataNotify_, MemType::DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
396 : }
397 : // Drain操作需读取对端srcmem
398 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
399 : exchangeDataPtr, exchangeDataBlankSize,
400 : reinterpret_cast<void*>(&memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)]), sizeof(MemMsg)));
401 0 : exchangeDataPtr += sizeof(MemMsg);
402 0 : exchangeDataBlankSize -= sizeof(MemMsg);
403 : }
404 0 : if (!isHybridMode_) {
405 0 : CHK_RET(ConstructExchangeDataForSend(exchangeDataPtr, exchangeDataBlankSize));
406 : }
407 0 : if (machinePara_.userMemEnable && UseMultiQp()) {
408 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
409 0 : std::shared_ptr<LocalIpcNotify> oneNotify;
410 0 : if (machinePara_.isAicpuModeEn) {
411 0 : CHK_RET(CreateNotifyBuffer(
412 : oneNotify, MemType::MULTI_QP_DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize,
413 : NotifyLoadType::DEVICE_NOTIFY));
414 : } else {
415 0 : CHK_RET(CreateNotifyBuffer(
416 : oneNotify, MemType::MULTI_QP_DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
417 : }
418 0 : multiQpDataNotify_.push_back(std::move(oneNotify));
419 0 : }
420 : }
421 :
422 : // 创建notify pool资源
423 : // 单qp创建1*notifyNum个，多qp创建qpNum*notifyNum个
424 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
425 0 : std::vector<std::shared_ptr<LocalIpcNotify>> notifyVec;
426 0 : CHK_RET(CreateNotifyVectorBuffer(notifyVec, exchangeDataPtr, exchangeDataBlankSize));
427 0 : userMultiQpLocalNotify_.push_back(std::move(notifyVec));
428 0 : }
429 :
430 0 : u8 localEnableAtomicWrite = machinePara_.enableAtomicWrite ? 1 : 0;
431 0 : sRet = memcpy_s(exchangeDataPtr, sizeof(u8), reinterpret_cast<void*>(&localEnableAtomicWrite), sizeof(u8));
432 0 : CHK_PRT_RET(
433 : sRet != EOK,
434 : HCCL_ERROR(
435 : "[Set][LocalMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu], cnt[%zu]",
436 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(bool), sizeof(bool)),
437 : HCCL_E_MEMORY);
438 0 : exchangeDataPtr += sizeof(u8);
439 0 : exchangeDataBlankSize -= sizeof(u8);
440 :
441 0 : if (machinePara_.isIndOp) {
442 0 : CHK_RET(RegCustomUserMem(exchangeDataPtr, exchangeDataBlankSize));
443 : }
444 :
445 0 : if (exchangeDataBlankSize != 0) {
446 0 : HCCL_ERROR(
447 : "[TransportIbverbs][ConstructExchangeForSend] failed to construct exchange Data "
448 : "exchangeDataBlankSize[%llu]",
449 : exchangeDataBlankSize);
450 0 : return HCCL_E_INTERNAL;
451 : }
452 :
453 0 : HCCL_DEBUG("[TransportIbverbs] ConstructExchangeForSend finished.");
454 0 : return HCCL_SUCCESS;
455 : }
456 :
457 0 : HcclResult TransportIbverbs::ParseReceivedExchangeData()
458 : {
459 0 : u8* exchangeDataPtr = exchangeDataForRecv_.data();
460 0 : u64 exchangeDataBlankSize = exchangeDataTotalSize_;
461 :
462 : // 首先解析qp对数量，并作一致性校验
463 0 : u32 localQpNum = UseMultiQp() ? qpsPerConnection_ : 1;
464 0 : u32 remoteQpNum = 0;
465 0 : s32 sRet = memcpy_s(reinterpret_cast<void*>(&remoteQpNum), sizeof(u32), exchangeDataPtr, sizeof(u32));
466 0 : CHK_PRT_RET(
467 : sRet != EOK,
468 : HCCL_ERROR(
469 : "[Get][RemoteMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
470 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
471 : HCCL_E_MEMORY);
472 0 : CHK_PRT_RET(
473 : localQpNum != remoteQpNum,
474 : HCCL_ERROR(
475 : "[TransportIbverbs][ParseReceivedExchangeData]"
476 : "local qps[%u] not equal to remote qps[%u], rank:local[%u],remote[%u]",
477 : localQpNum, remoteQpNum, machinePara_.localUserrank, machinePara_.remoteUserrank),
478 : HCCL_E_INTERNAL);
479 0 : exchangeDataPtr += sizeof(u32);
480 0 : exchangeDataBlankSize -= sizeof(u32);
481 :
482 0 : if (UseMultiQp()) {
483 : // 再解析multiQpThreshold，并作一致性校验
484 0 : u32 localmultiQpThreshold = GetExternalInputMultiQpThreshold();
485 0 : u32 remotemultiQpThreshold = 0;
486 0 : sRet = memcpy_s(reinterpret_cast<void*>(&remotemultiQpThreshold), sizeof(u32), exchangeDataPtr, sizeof(u32));
487 0 : CHK_PRT_RET(
488 : sRet != EOK,
489 : HCCL_ERROR(
490 : "[Get][RemoteMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
491 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
492 : HCCL_E_MEMORY);
493 0 : CHK_PRT_RET(
494 : localmultiQpThreshold != remotemultiQpThreshold,
495 : HCCL_ERROR(
496 : "[TransportIbverbs][ParseReceivedExchangeData]"
497 : "local env HCCL_MULTI_QP_THRESHOLD[%u] not equal to remote env HCCL_MULTI_QP_THRESHOLD[%u], "
498 : "rank:local[%u],remote[%u]",
499 : localmultiQpThreshold, remotemultiQpThreshold, machinePara_.localUserrank, machinePara_.remoteUserrank),
500 : HCCL_E_INTERNAL);
501 0 : exchangeDataPtr += sizeof(u32);
502 0 : exchangeDataBlankSize -= sizeof(u32);
503 : }
504 :
505 0 : if (machinePara_.userMemEnable) {
506 0 : CHK_RET(GetRemoteAddr(MemType::USER_OUTPUT_MEM, exchangeDataPtr, exchangeDataBlankSize));
507 0 : CHK_RET(GetRemoteAddr(MemType::USER_INPUT_MEM, exchangeDataPtr, exchangeDataBlankSize));
508 0 : CHK_RET(GetRemoteAddr(MemType::DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
509 0 : s32 sret = memcpy_s(
510 0 : &remoteDataNotifyMsg_, sizeof(MemMsg), &remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)],
511 : sizeof(MemMsg));
512 0 : CHK_PRT_RET(
513 : sret != EOK,
514 : HCCL_ERROR(
515 : "[Get][RemoteMem]errNo[0x%016llx] In lbv exp init, memory copy failed. errorno[%d], "
516 : "params:destMaxSize[%zu],count[%zu]",
517 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sret, sizeof(MemMsg), sizeof(MemMsg)),
518 : HCCL_E_MEMORY);
519 0 : CHK_RET(GetRemoteAddr(MemType::ACK_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
520 0 : CHK_RET(GetRemoteAddr(MemType::DATA_ACK_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
521 : }
522 0 : if (!machinePara_.userMemEnable && machinePara_.drainEnable) {
523 0 : CHK_RET(GetRemoteAddr(MemType::DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
524 0 : CHK_RET(GetRemoteAddr(MemType::NOTIFY_SRC_MEM, exchangeDataPtr, exchangeDataBlankSize));
525 : }
526 :
527 0 : if (!isHybridMode_) {
528 0 : CHK_RET(ParseExchangeData(exchangeDataPtr, exchangeDataBlankSize));
529 : }
530 :
531 0 : if (machinePara_.userMemEnable && UseMultiQp()) {
532 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
533 0 : CHK_RET(GetRemoteAddr(MemType::MULTI_QP_DATA_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize));
534 : }
535 : }
536 :
537 : // 解析远端新增的notify资源，二维vec大小：qpNum*notifyNum
538 0 : userMultiQpRemoteNotifyMsg_.resize(qpsPerConnection_);
539 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
540 0 : userMultiQpRemoteNotifyMsg_[i].resize(notifyNum_);
541 0 : for (u32 j = 0; j < notifyNum_; j++) {
542 0 : CHK_RET(GetRemoteNotifyAddr(exchangeDataPtr, exchangeDataBlankSize, userMultiQpRemoteNotifyMsg_[i][j]));
543 : }
544 : }
545 :
546 : // 解析远端是否都支持atomic write，仅本端和对端都支持时才使能atomic write
547 0 : u8 remoteEnableAtomicWrite = 0;
548 0 : sRet = memcpy_s(reinterpret_cast<void*>(&remoteEnableAtomicWrite), sizeof(u8), exchangeDataPtr, sizeof(u8));
549 0 : CHK_PRT_RET(
550 : sRet != EOK,
551 : HCCL_ERROR(
552 : "[Get][RemoteMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu], cnt[%zu]",
553 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(bool), sizeof(bool)),
554 : HCCL_E_MEMORY);
555 0 : useAtomicWrite_ = machinePara_.enableAtomicWrite && (remoteEnableAtomicWrite != 0);
556 0 : HCCL_INFO(
557 : "atomic write enable only when both the local and remote support, local[%d], remote[%d], result[%d]",
558 : machinePara_.enableAtomicWrite, remoteEnableAtomicWrite, useAtomicWrite_);
559 0 : exchangeDataPtr += sizeof(u8);
560 0 : exchangeDataBlankSize -= sizeof(u8);
561 :
562 0 : if (machinePara_.isIndOp) {
563 0 : CHK_RET(GetIndOpRemoteAddr(exchangeDataPtr, exchangeDataBlankSize));
564 : }
565 :
566 0 : if (exchangeDataBlankSize != 0) {
567 0 : HCCL_ERROR(
568 : "[TransportIbverbs][ParseReceivedExchangeData] failed to Parse exchange Data "
569 : "exchangeDataBlankSize[%llu]",
570 : exchangeDataBlankSize);
571 0 : return HCCL_E_INTERNAL;
572 : }
573 0 : HCCL_DEBUG("Parse Received ExchangeData success!");
574 0 : return HCCL_SUCCESS;
575 : }
576 :
577 0 : void TransportIbverbs::ModifyAtomicWriteAfterReduce(u32& preWrOpcode, u64 wqeType, u32& opcode, u32& immData)
578 : {
579 0 : bool isNotifyWqe = wqeType == static_cast<u64>(WqeType::WQE_TYPE_DATA_NOTIFY)
580 0 : || wqeType == static_cast<u64>(WqeType::WQE_TYPE_ACK_NOTIFY)
581 0 : || wqeType == static_cast<u64>(WqeType::WQE_TYPE_DATA_ACK_NOTIFY);
582 0 : if (useAtomicWrite_ && preWrOpcode == RA_WR_RDMA_REDUCE_WRITE && isNotifyWqe) {
583 0 : opcode = RA_WR_RDMA_ATOMIC_WRITE;
584 0 : immData = machinePara_.isAicpuModeEn ? htobe32(0x1) : 0x1; // aicpu展开时，HCCL直调RoCE驱动，需进行字节序转换
585 : }
586 0 : HCCL_DEBUG(
587 : "%s preWrOpcode[%u] useAtomicWrite[%d] wqeType[%d] opcode[0x%x] immdata[%u]", __func__, preWrOpcode,
588 : useAtomicWrite_, wqeType, opcode, immData);
589 0 : preWrOpcode = opcode;
590 0 : }
591 :
592 0 : u32 TransportIbverbs::GetQpsPerConnection()
593 : {
594 0 : u32 externalQps = std::max(static_cast<u32>(machinePara_.srcPorts.size()), 1U);
595 0 : s32 qpMode = GetQpMode();
596 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
597 0 : && externalQps != HCCL_QPS_PER_CONNECTION_DEFAULT) {
598 0 : HCCL_RUN_INFO(
599 : "HCCL_RDMA_QPS_PER_CONNECTION is set to [%u] but it is not effective in offline mode.", externalQps);
600 0 : } else if (qpMode != OPBASE_QP_MODE_EXT && externalQps > 1) {
601 0 : HCCL_RUN_INFO(
602 : "HCCL_RDMA_QPS_PER_CONNECTION is set to [%u] but current devType[%d] does not support multi-QP.",
603 : externalQps, machinePara_.deviceType);
604 0 : return 1; // 非单算子模式仅支持单QP， QPS = 1
605 : }
606 0 : if (GetWorkflowMode() == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
607 0 : return externalQps; // only work for opbase mode
608 : }
609 0 : return 1; // 非单算子模式仅支持单QP， QPS = 1
610 : }
611 :
612 0 : HcclResult TransportIbverbs::GetNicHandle()
613 : {
614 0 : RaResourceInfo raResourceInfo;
615 0 : CHK_RET(NetworkManager::GetInstance(machinePara_.deviceLogicId).GetRaResourceInfo(raResourceInfo));
616 0 : std::map<HcclIpAddress, IpSocket>& tmpSocketMap = machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE ?
617 : raResourceInfo.nicSocketMap :
618 : raResourceInfo.hostNetSocketMap;
619 :
620 0 : HcclIpAddress localIpAddr = machinePara_.localIpAddr;
621 :
622 : // 获取 nicRdmaHandle
623 0 : auto itSocket = tmpSocketMap.find(localIpAddr);
624 0 : if (itSocket == tmpSocketMap.end()) {
625 0 : HCCL_ERROR(
626 : "[Get][NicHandle]In get nic handle, can not find socket handle, handle size[%u], "
627 : "local ip[%s]",
628 : tmpSocketMap.size(), localIpAddr.GetReadableAddress());
629 0 : return HCCL_E_PARA;
630 : }
631 :
632 0 : nicRdmaHandle_ = itSocket->second.nicRdmaHandle;
633 0 : CHK_PTR_NULL(nicRdmaHandle_);
634 :
635 0 : return HCCL_SUCCESS;
636 0 : }
637 :
638 0 : inline static void MultiQpAdjustQpCapacity(struct QpExtAttrs& attrs)
639 : {
640 0 : constexpr int multiQpCapacityRatio = 2;
641 0 : attrs.qpAttr.cap.max_send_wr /= multiQpCapacityRatio;
642 0 : attrs.cqAttr.sendCqDepth /= multiQpCapacityRatio;
643 0 : }
644 :
645 : // 创建一个QP
646 0 : HcclResult TransportIbverbs::CreateOneQp(
647 : s32 qpMode, u32 qpsPerConnection, QpHandle& qpHandle, AiQpInfo& aiQpInfo, bool useAicpu, u32 udpSport)
648 : {
649 0 : bool isUseQpCreateWithAttrs = false;
650 0 : CHK_RET(IsUseQpCreateWithAttrs(isUseQpCreateWithAttrs, qpMode));
651 : HcclResult ret;
652 0 : std::string useAicpuTitle = useAicpu ? std::string("aicpu ") : std::string("");
653 0 : std::string qpInfo = useAicpuTitle + std::string("rank:") + std::to_string(machinePara_.localWorldRank)
654 0 : + std::string(",localUserrank:") + std::to_string(machinePara_.localUserrank)
655 0 : + std::string(",localIpAddr: ") + std::string(machinePara_.localIpAddr.GetReadableAddress())
656 0 : + std::string(",deviceLogicId:") + std::to_string(machinePara_.deviceLogicId);
657 0 : struct QpExtAttrs attrs {};
658 : // 判断是否为NORMALQP需要使用qpMode_; hostnic场景的qpmode也是NORMALQP
659 0 : if (useAicpu || qpMode_ == QPMode::NORMAL) {
660 0 : bool isWorkFlowLib = (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB);
661 0 : CHK_RET(ConstructQpAttrs(qpMode, attrs, machinePara_.queueDepthAttr, isWorkFlowLib));
662 0 : bool isA3Aicpu = machinePara_.isAicpuModeEn && (machinePara_.deviceType == DevType::DEV_TYPE_910_93);
663 0 : if (isA3Aicpu && workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
664 0 : && qpMode == OPBASE_QP_MODE_EXT) {
665 0 : attrs.qpAttr.cap.max_send_wr = machinePara_.queueDepthAttr.sqDepth == INVALID_UINT ?
666 : AICPU_SQ_CQ_DEPTH :
667 0 : attrs.qpAttr.cap.max_send_wr;
668 0 : attrs.cqAttr.sendCqDepth = machinePara_.queueDepthAttr.sendCqDepth == INVALID_UINT ?
669 : AICPU_SQ_CQ_DEPTH :
670 : attrs.cqAttr.sendCqDepth;
671 : }
672 : // A3 aicpu图模式使用单个qp, qp深度为socket数量*128
673 0 : bool isAicpuLib = machinePara_.isAicpuModeEn && (machinePara_.deviceType == DevType::DEV_TYPE_910_93)
674 0 : && (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB);
675 0 : if (!UseMultiQp() && isAicpuLib) {
676 0 : attrs.qpAttr.cap.max_send_wr = machinePara_.sockets.size() * DEFAULT_OFFLINE_MAX_SEND_WR;
677 0 : } else if (UseMultiQp()) {
678 0 : MultiQpAdjustQpCapacity(attrs);
679 : }
680 0 : HCCL_DEBUG(
681 : "qp set max_send_wr %u, socket size %u, isWorkFlowLib %d", attrs.qpAttr.cap.max_send_wr,
682 : machinePara_.sockets.size(), isWorkFlowLib);
683 :
684 0 : attrs.udpSport = udpSport;
685 0 : ret = hrtRaAiQpCreate(machinePara_.localDeviceId, nicRdmaHandle_, &attrs, &aiQpInfo, qpHandle);
686 0 : HCCL_DEBUG("aiQpAddr:%llu db_index:%u, sq_index=%u", aiQpInfo.aiQpAddr, aiQpInfo.dbIndex, aiQpInfo.sqIndex);
687 0 : qpInfo = qpInfo + std::string(",sendCqDepth:") + std::to_string(attrs.cqAttr.sendCqDepth);
688 0 : } else if (!isUseQpCreateWithAttrs && qpsPerConnection == HCCL_QPS_PER_CONNECTION_DEFAULT) {
689 0 : ret = HrtRaQpCreate(nicRdmaHandle_, QP_FLAG_RC, qpMode, qpHandle);
690 0 : } else if (!isUseQpCreateWithAttrs && qpsPerConnection != HCCL_QPS_PER_CONNECTION_DEFAULT) {
691 0 : HCCL_ERROR("qpsPerConnection[%u] is set but qpMode[%d] is not supported", qpsPerConnection, qpMode);
692 0 : return HCCL_E_PARA;
693 : } else {
694 0 : CHK_RET(ConstructQpAttrs(qpMode, attrs, machinePara_.queueDepthAttr));
695 0 : if (machinePara_.deviceType == DevType::DEV_TYPE_910_93 && !machinePara_.isAicpuModeEn
696 0 : && workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE && qpMode == OPBASE_QP_MODE_EXT) {
697 : attrs.qpAttr.cap.max_send_wr
698 0 : = machinePara_.queueDepthAttr.sqDepth == INVALID_UINT ? HOST_SQ_CQ_DEPTH : attrs.qpAttr.cap.max_send_wr;
699 : attrs.cqAttr.sendCqDepth
700 0 : = machinePara_.queueDepthAttr.sendCqDepth == INVALID_UINT ? HOST_SQ_CQ_DEPTH : attrs.cqAttr.sendCqDepth;
701 : }
702 0 : if (UseMultiQp()) {
703 0 : MultiQpAdjustQpCapacity(attrs);
704 : }
705 0 : attrs.udpSport = udpSport;
706 0 : ret = hrtRaQpCreateWithAttrs(nicRdmaHandle_, &attrs, qpHandle);
707 0 : qpInfo = qpInfo + std::string(",sendCqDepth:") + std::to_string(attrs.cqAttr.sendCqDepth);
708 : }
709 :
710 0 : RPT_ENV_ERR(
711 : ret != 0 || (qpHandle == nullptr), "EI0007", vector<string>({"resource_type", "resource_info"}),
712 : vector<string>({"qp", "CreateOneQp"}));
713 :
714 0 : CHK_PRT_RET(
715 : ret != HCCL_SUCCESS,
716 : HCCL_ERROR(
717 : "[%s][%s]create qp failed, localDeviceId[%d], qpMode[%d]", LOG_KEYWORDS_INIT_GROUP.c_str(),
718 : LOG_KEYWORDS_RESOURCE.c_str(), machinePara_.localDeviceId, qpMode),
719 : HCCL_E_ROCE_CONNECT);
720 :
721 : // 表示没有通过config配置，则使用环境变量配置
722 0 : CHK_RET(SetQpAttrQos(qpHandle, machinePara_.tc, machinePara_.sl));
723 : // 配置RDMA Timeout时间
724 0 : CHK_RET(SetQpAttrTimeOut(qpHandle, machinePara_.retryInterval));
725 : // 配置RDMA Retry Cnt重传次数
726 0 : CHK_RET(SetQpAttrRetryCnt(qpHandle, machinePara_.retryCnt));
727 : // qpn map 插入
728 0 : struct QpAttr attr {};
729 0 : CHK_RET(hrtRaGetQpAttr(qpHandle, &attr));
730 :
731 0 : g_qpn2IbversLinkMap_.Emplace(((static_cast<u64>(machinePara_.localDeviceId) << DEV_PHY_ID_BIT) | attr.qpn), this);
732 :
733 0 : HCCL_DEBUG("ra qp create success, use input udpSport[%u].", udpSport);
734 0 : return HCCL_SUCCESS;
735 0 : }
736 :
737 0 : HcclResult TransportIbverbs::CreateSingleQp(s32 qpMode) // 根据socket个数创建QP（下沉模板不够用多QP）
738 : {
739 0 : u32 socketNum = 1;
740 : // A3 aicpu图模式只使用1个qp，qp深度为socketNum*128，最大不超过32K
741 0 : bool isAicpuLib = machinePara_.isAicpuModeEn && (machinePara_.deviceType == DevType::DEV_TYPE_910_93)
742 0 : && (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OPS_KERNEL_INFO_LIB);
743 0 : if (!UseMultiQp() && !isAicpuLib) {
744 0 : socketNum = machinePara_.sockets.size();
745 : }
746 : // 原来是 machinePara_.socketFdHandles 换成 machinePara_.sockets
747 0 : for (u32 i = 0; i < socketNum; i++) {
748 0 : QpHandle qpHandle = nullptr;
749 0 : u32 udpSport = machinePara_.srcPorts.empty() ? 0 : machinePara_.srcPorts[0];
750 0 : CHK_RET(CreateOneQp(
751 : qpMode, HCCL_QPS_PER_CONNECTION_DEFAULT, qpHandle, combineAiQpInfo_.aiQpInfo, machinePara_.isAicpuModeEn,
752 : udpSport));
753 0 : CombineQpHandle tmpCombineQpHandle;
754 0 : tmpCombineQpHandle.qpHandle = qpHandle;
755 0 : combineQpHandles_.push_back(tmpCombineQpHandle);
756 : }
757 0 : return HCCL_SUCCESS;
758 : }
759 :
760 0 : HcclResult TransportIbverbs::CreateMultiQp(s32 qpMode, u32 qpsPerConnection)
761 : {
762 : // 配置了多qp源端口号时的处理流程
763 0 : if (machinePara_.srcPorts.size() > 0) {
764 0 : HCCL_DEBUG("[TransportIbverbs][CreateMultiQp]use Multi qp create qps.");
765 : // 创建qp
766 0 : for (const auto& port : machinePara_.srcPorts) {
767 0 : QpHandle qpHandle = nullptr;
768 0 : AiQpInfo tmpAiQpInfo{};
769 0 : CHK_RET(CreateOneQp(qpMode, qpsPerConnection, qpHandle, tmpAiQpInfo, machinePara_.isAicpuModeEn, port));
770 0 : multiCombineQpHandles_.push_back(CombineQpHandle(qpHandle));
771 0 : combineAiQpInfos_.push_back(CombineQpInfo(tmpAiQpInfo));
772 : }
773 : }
774 0 : HCCL_DEBUG("ra multi-qp creation success.");
775 0 : return HCCL_SUCCESS;
776 : }
777 :
778 12 : s32 TransportIbverbs::GetQpMode()
779 : {
780 12 : s32 qpMode = NORMAL_QP_MODE;
781 :
782 12 : if (qpMode_ == QPMode::NORMAL) {
783 0 : return qpMode;
784 : }
785 :
786 12 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
787 12 : if (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
788 0 : qpMode = (machinePara_.deviceType == DevType::DEV_TYPE_910B
789 0 : || machinePara_.deviceType == DevType::DEV_TYPE_910_93) ?
790 : OPBASE_QP_MODE_EXT :
791 : OPBASE_QP_MODE;
792 : // isCapture需要创建下沉QP
793 0 : qpMode = (qpMode == OPBASE_QP_MODE_EXT && qpMode_ == QPMode::OFFLOAD) ? OFFLINE_QP_MODE_EXT : qpMode;
794 0 : isCapture_ = (qpMode == OFFLINE_QP_MODE_EXT) ? true : false;
795 : } else {
796 12 : qpMode = (machinePara_.deviceType == DevType::DEV_TYPE_910B
797 12 : || machinePara_.deviceType == DevType::DEV_TYPE_910_93) ?
798 : OFFLINE_QP_MODE_EXT :
799 : OFFLINE_QP_MODE;
800 : }
801 : }
802 12 : if (machinePara_.isAicpuModeEn) {
803 3 : qpMode = (machinePara_.deviceType == DevType::DEV_TYPE_910B
804 3 : || machinePara_.deviceType == DevType::DEV_TYPE_910_93) ?
805 : OPBASE_QP_MODE_EXT :
806 : OPBASE_QP_MODE;
807 : }
808 12 : return qpMode;
809 : }
810 :
811 12 : bool TransportIbverbs::UseMultiQp()
812 : {
813 12 : s32 qpMode = GetQpMode();
814 12 : HCCL_DEBUG("[TransportIbverbs]UseMultiQp qpsPerConnection[%u]", qpsPerConnection_);
815 12 : if (qpMode == OPBASE_QP_MODE_EXT && qpsPerConnection_ > 1) {
816 3 : return true;
817 : }
818 9 : return false;
819 : }
820 :
821 0 : HcclResult TransportIbverbs::CreateQp()
822 : {
823 0 : s32 qpMode = GetQpMode();
824 0 : HCCL_DEBUG("[TransportIbverbs][CreateQp] QpMode[%u]", qpMode);
825 0 : CHK_RET(CreateSingleQp(qpMode));
826 0 : if (UseMultiQp()) {
827 0 : HCCL_DEBUG("[TransportIbverbs]Create MultiQP begin");
828 0 : CHK_RET(CreateMultiQp(qpMode, qpsPerConnection_));
829 : }
830 0 : HCCL_DEBUG("ra qp create %u qp success.", combineQpHandles_.size());
831 0 : return HCCL_SUCCESS;
832 : }
833 :
834 0 : HcclResult TransportIbverbs::InitQpConnect()
835 : {
836 : /* 创建QP操作句柄 */
837 0 : qpsPerConnection_ = GetQpsPerConnection();
838 :
839 0 : CHK_RET(ExchangeCapabilityHybrid());
840 :
841 0 : CHK_RET(CreateQp());
842 :
843 0 : CHK_RET(FillExchangeDataTotalSize());
844 :
845 : // 注册notify 内存信息
846 0 : CHK_RET(CreateNotifyValueBuffer());
847 :
848 0 : CHK_RET(ConstructExchangeForSend());
849 0 : HCCL_DEBUG("[TransportIbverbs] resource create done exchangeDataTotalSize_[%llu]", exchangeDataTotalSize_);
850 :
851 0 : HcclResult ret = defaultSocket_->Send(exchangeDataForSend_.data(), exchangeDataTotalSize_);
852 0 : CHK_PRT_RET(
853 : ret != HCCL_SUCCESS,
854 : HCCL_ERROR(
855 : "[TransportIbverbs][InitQpConnect] failed to send exchangeData exchangeDataTotalSize[%llu], "
856 : "custom exchange data size [%llu].",
857 : exchangeDataTotalSize_, machinePara_.exchangeInfo.size()),
858 : ret);
859 0 : HCCL_DEBUG("[TransportIbverbs]Seocket Send finished, exchangeDataTotalSize[%llu]", exchangeDataTotalSize_);
860 :
861 0 : exchangeDataForRecv_.resize(exchangeDataTotalSize_);
862 0 : ret = defaultSocket_->Recv(exchangeDataForRecv_.data(), exchangeDataTotalSize_);
863 0 : CHK_PRT_RET(
864 : ret != HCCL_SUCCESS,
865 : HCCL_ERROR(
866 : "[TransportIbverbs][InitQpConnect] failed to recv exchangeData exchangeDataTotalSize[%llu], "
867 : "custom exchange data size [%llu].",
868 : exchangeDataTotalSize_, machinePara_.exchangeInfo.size()),
869 : ret);
870 :
871 0 : HCCL_DEBUG("[TransportIbverbs][Init] Socket Data Recved");
872 :
873 0 : CHK_RET(ParseReceivedExchangeData());
874 :
875 : // 连接Qp
876 0 : CHK_RET(ConnectQp());
877 0 : HCCL_INFO("In link ibv, qp status has ready");
878 0 : return HCCL_SUCCESS;
879 : }
880 :
881 0 : HcclResult TransportIbverbs::ConnectSingleQp(std::function<bool()> needStop)
882 : {
883 : // QP建链
884 0 : for (u32 i = 0; i < combineQpHandles_.size(); i++) {
885 0 : CHK_RET(HrtRaQpConnectAsync(combineQpHandles_[i].qpHandle, machinePara_.sockets[i]->GetFdHandle(), needStop));
886 : }
887 : // 查询QP建链是否成功
888 0 : s32 qpStatus = 0;
889 0 : s32 raRet = 0;
890 0 : auto startTime = std::chrono::steady_clock::now();
891 0 : HCCL_INFO("In link ibv, waiting for qp status ready...");
892 0 : for (u32 i = 0; i < combineQpHandles_.size(); i++) {
893 : while (true) {
894 0 : CHK_PRT_RET(needStop(), HCCL_ERROR("Terminating operation due to external request"), HCCL_E_INTERNAL);
895 :
896 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout_) {
897 0 : HCCL_ERROR("[Connect][Qp]get qp status timeout_=[%lld ms], qp_status=[%d]", timeout_, qpStatus);
898 0 : return HCCL_E_TIMEOUT;
899 : }
900 0 : raRet = hrtGetRaQpStatus(combineQpHandles_[i].qpHandle, &qpStatus);
901 0 : if ((!raRet) && (qpStatus == 1)) { // 为1时，qp 建链成功
902 0 : HCCL_INFO("In link ibv, %u of %u QP get status success.", (i + 1), combineQpHandles_.size());
903 0 : break;
904 : } else {
905 : // qp建链需要时间，获取qp状态直至超时
906 0 : SaluSleep(WAIT_US_COUNT);
907 : }
908 : }
909 : }
910 0 : return HCCL_SUCCESS;
911 : }
912 :
913 0 : HcclResult TransportIbverbs::ConnectMultiQp(u32 qpsPerConnection, std::function<bool()> needStop)
914 : {
915 : // 多QP下，复用同一个socket handle来modify QP, 此时需要串行创建
916 0 : if (machinePara_.sockets.size() != 2) { // 2：多QP下需要一个额外的Socket来做QP状态迁移同步
917 0 : return HCCL_E_INTERNAL;
918 : }
919 0 : for (u32 i = 0; i < qpsPerConnection; i++) {
920 : // QP建链
921 0 : u8 localQpConnectReady = 1;
922 0 : u8 remoteQpConnectReady = 0;
923 0 : CHK_RET(machinePara_.sockets[1]->Send(&localQpConnectReady, 1));
924 0 : CHK_RET(machinePara_.sockets[1]->Recv(&remoteQpConnectReady, 1));
925 0 : std::string aicpu = machinePara_.isAicpuModeEn ? "aicpu" : "";
926 0 : CHK_PRT_RET(
927 : remoteQpConnectReady != localQpConnectReady,
928 : HCCL_ERROR(
929 : "[TransportIbverbs] %s multi Qp Connected checking failed! %u of %u QP", aicpu.c_str(), (i + 1),
930 : qpsPerConnection),
931 : HCCL_E_NETWORK);
932 0 : CHK_RET(
933 : HrtRaQpConnectAsync(multiCombineQpHandles_[i].qpHandle, machinePara_.sockets[0]->GetFdHandle(), needStop));
934 : // 查询QP建链是否成功
935 0 : s32 qpStatus = 0;
936 0 : s32 raRet = 0;
937 0 : auto startTime = std::chrono::steady_clock::now();
938 0 : HCCL_INFO(
939 : "In link ibv, waiting for qp status ready... %u of %u %s QP", (i + 1), multiCombineQpHandles_.size(),
940 : aicpu.c_str());
941 : while (true) {
942 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout_) {
943 0 : HCCL_ERROR(
944 : "[Connect][Qp]get qp status timeout_=[%lld ms], qp_status=[%d], index[%u]", timeout_, qpStatus, i);
945 0 : return HCCL_E_TIMEOUT;
946 : }
947 0 : raRet = hrtGetRaQpStatus(multiCombineQpHandles_[i].qpHandle, &qpStatus);
948 0 : if ((!raRet) && (qpStatus == 1)) { // 为1时，qp 建链成功
949 0 : HCCL_INFO(
950 : "In link ibv, %u of %u %s QP get status success.", (i + 1), multiCombineQpHandles_.size(),
951 : aicpu.c_str());
952 0 : break;
953 : } else {
954 : // qp建链需要时间，获取qp状态直至超时
955 0 : SaluSleep(WAIT_US_COUNT);
956 : }
957 : }
958 0 : }
959 0 : return HCCL_SUCCESS;
960 : }
961 :
962 0 : HcclResult TransportIbverbs::ConnectQp()
963 : {
964 0 : CHK_RET(ConnectSingleQp([this]() -> bool {
965 : return this->GetStopFlag();
966 : }));
967 0 : if (UseMultiQp()) {
968 0 : CHK_RET(ConnectMultiQp(qpsPerConnection_, [this]() -> bool {
969 : return this->GetStopFlag();
970 : }));
971 : }
972 0 : return HCCL_SUCCESS;
973 : }
974 :
975 1 : HcclResult TransportIbverbs::Fence()
976 : {
977 1 : fence_ = true;
978 1 : return HCCL_SUCCESS;
979 : }
980 :
981 1 : HcclResult TransportIbverbs::AddWqeList(
982 : void* dstMemPtr, const void* srcMemPtr, u64 srcMemSize, WqeType wqeType, WrAuxInfo& aux,
983 : std::vector<WqeInfo>& wqeInfoVec)
984 : {
985 1 : WqeInfo wqeInfoTmp;
986 :
987 1 : wqeInfoTmp.wqeData.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(dstMemPtr));
988 1 : wqeInfoTmp.wqeData.sendFlags = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
989 1 : fence_ = false;
990 1 : wqeInfoTmp.wqeData.memList.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
991 1 : wqeInfoTmp.wqeData.memList.len = srcMemSize;
992 1 : wqeInfoTmp.wqeData.memList.lkey = 0;
993 :
994 1 : switch (wqeType) {
995 0 : case WqeType::WQE_TYPE_DATA:
996 : case WqeType::WQE_TYPE_DATA_NOTIFY:
997 : case WqeType::WQE_TYPE_ACK_NOTIFY:
998 : case WqeType::WQE_TYPE_DATA_ACK_NOTIFY:
999 : case WqeType::WQE_TYPE_DATA_WITH_NOTIFY:
1000 0 : wqeInfoTmp.wqeData.op = RA_WR_RDMA_WRITE;
1001 0 : wqeInfoTmp.wqeType = static_cast<u64>(wqeType);
1002 0 : break;
1003 0 : case WqeType::WQE_TYPE_DATA_WITH_REDUCE:
1004 0 : wqeInfoTmp.wqeData.op = RA_WR_RDMA_REDUCE_WRITE;
1005 0 : wqeInfoTmp.wqeData.aux = aux;
1006 : // REDUCE WRITE 作为特殊的DATA
1007 0 : wqeInfoTmp.wqeType = static_cast<u64>(WqeType::WQE_TYPE_DATA);
1008 0 : break;
1009 1 : case WqeType::WQE_TYPE_READ_DATA:
1010 1 : wqeInfoTmp.wqeData.op = RA_WR_RDMA_READ;
1011 1 : wqeInfoTmp.wqeType = static_cast<u64>(wqeType);
1012 1 : break;
1013 0 : default:
1014 0 : HCCL_ERROR("error wqeType[%d]", wqeType);
1015 0 : return HCCL_E_INTERNAL;
1016 : }
1017 1 : CHK_RET(GetWqeDataOffsetAndNotifyId(wqeType, wqeInfoTmp.wqeDataOffset, wqeInfoTmp.notifyId));
1018 :
1019 1 : wqeInfoVec.push_back(wqeInfoTmp);
1020 1 : return HCCL_SUCCESS;
1021 : }
1022 :
1023 0 : HcclResult TransportIbverbs::ConstructPayLoadWqe(
1024 : void* dstMemPtr, const void* src, u64 len, WqeType wqeType, WrAuxInfo& aux, std::vector<WqeInfo>& wqeInfoVec,
1025 : u32 txSendDataTimes)
1026 : {
1027 : HcclResult ret;
1028 : // 发送数据Wqe
1029 0 : for (u32 txSendDataIdx = 0; txSendDataIdx < txSendDataTimes; txSendDataIdx++) {
1030 0 : u64 txSendDataOffset = txSendDataIdx * RDMA_SEND_MAX_SIZE;
1031 0 : u64 txSendDataSize = (txSendDataIdx == (txSendDataTimes - 1)) ? len - txSendDataOffset : RDMA_SEND_MAX_SIZE;
1032 :
1033 0 : void* txdstMemPtr = reinterpret_cast<void*>(reinterpret_cast<char*>(dstMemPtr) + txSendDataOffset);
1034 :
1035 0 : const void* txsrcMemPtr = reinterpret_cast<const void*>(reinterpret_cast<const char*>(src) + txSendDataOffset);
1036 0 : ret = AddWqeList(txdstMemPtr, txsrcMemPtr, txSendDataSize, wqeType, aux, wqeInfoVec);
1037 0 : CHK_PRT_RET(
1038 : ret != HCCL_SUCCESS,
1039 : HCCL_ERROR(
1040 : "[TransportIbverbs][TxAsync]errNo[0x%016llx] In lbv exp, add wqe list failed."
1041 : "srcMemSize[%llu Byte]",
1042 : HCCL_ERROR_CODE(ret), txSendDataSize),
1043 : ret);
1044 : }
1045 :
1046 0 : return HCCL_SUCCESS;
1047 : }
1048 :
1049 0 : HcclResult TransportIbverbs::TxPayLoad(
1050 : UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, WqeType wqeType, WrAuxInfo& aux,
1051 : std::vector<WqeInfo>& wqeInfoVec)
1052 : {
1053 0 : void* dstMemPtr = nullptr;
1054 0 : u64 dstMemSize = 0;
1055 : // 为保证单算子下不同数据量下子图的结构相同，zero byte message 时也需要下发task
1056 0 : u32 txSendDataTimes = (len == 0) ? 1 : (len + RDMA_SEND_MAX_SIZE - 1) / RDMA_SEND_MAX_SIZE;
1057 0 : CHK_RET(GetMemInfo(dstMemType, &dstMemPtr, &dstMemSize));
1058 :
1059 0 : if (dstOffset > dstMemSize) {
1060 0 : HCCL_ERROR(
1061 : "[TransportIbverbs][TxAsync]dst_mem_type=%d, dst_mem_ptr=%p, dst_offset=%llu, dst_mem_size=%llu Byte",
1062 : dstMemType, dstMemPtr, dstOffset, dstMemSize);
1063 0 : return HCCL_E_INTERNAL;
1064 : }
1065 :
1066 0 : dstMemPtr = reinterpret_cast<void*>(reinterpret_cast<char*>(dstMemPtr) + dstOffset);
1067 0 : CHK_RET(ConstructPayLoadWqe(dstMemPtr, src, len, wqeType, aux, wqeInfoVec, txSendDataTimes));
1068 :
1069 0 : return HCCL_SUCCESS;
1070 : }
1071 :
1072 0 : HcclResult TransportIbverbs::TxAsync(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
1073 : {
1074 0 : std::vector<WqeInfo> wqeInfoVec;
1075 0 : wqeInfoVec.reserve(WQE_RESERVE_LENGTH);
1076 0 : struct WrAuxInfo aux = {};
1077 0 : HCCL_DEBUG("TX src[%p] len[%llu] dstOffset[%llu]", src, len, dstOffset);
1078 :
1079 0 : if (src != nullptr) {
1080 0 : CHK_RET(TxPayLoad(dstMemType, dstOffset, src, len, WqeType::WQE_TYPE_DATA, aux, wqeInfoVec));
1081 : }
1082 :
1083 0 : CHK_RET(TxSendDataAndNotify(wqeInfoVec, stream, GetUseOneDoorbellValue()));
1084 0 : return HCCL_SUCCESS;
1085 0 : }
1086 :
1087 0 : HcclResult TransportIbverbs::TxWithReduce(
1088 : UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, const HcclDataType datatype, HcclReduceOp redOp,
1089 : Stream& stream)
1090 : {
1091 0 : std::vector<WqeInfo> wqeInfoVec;
1092 0 : wqeInfoVec.reserve(WQE_RESERVE_LENGTH);
1093 0 : struct WrAuxInfo aux = {};
1094 0 : aux.dataType = RDMA_REDUCE_DATA_TYPE_TABLE[datatype];
1095 0 : aux.reduceType = RDMA_REDUCE_OP_TYPE_TABLE[redOp];
1096 0 : if (aux.dataType == static_cast<uint8_t>(RdmaReduceDataType::RDMA_REDUCE_DATA_INVALID)
1097 0 : || aux.reduceType == static_cast<uint8_t>(RdmaReduceOpType::RDMA_REDUCE_OP_INVALID)) {
1098 0 : HCCL_ERROR(
1099 : "unsupported data type [%s] or Reduce type [%s]", GetDataTypeEnumStr(datatype).c_str(),
1100 : GetReduceOpEnumStr(redOp).c_str());
1101 0 : return HCCL_E_INTERNAL;
1102 : }
1103 :
1104 0 : CHK_PTR_NULL(src);
1105 0 : CHK_RET(TxPayLoad(dstMemType, dstOffset, src, len, WqeType::WQE_TYPE_DATA_WITH_REDUCE, aux, wqeInfoVec));
1106 :
1107 0 : CHK_RET(TxSendDataAndNotify(wqeInfoVec, stream, GetUseOneDoorbellValue()));
1108 0 : return HCCL_SUCCESS;
1109 0 : }
1110 :
1111 0 : HcclResult TransportIbverbs::TxWithReduce(
1112 : const std::vector<TxMemoryInfo>& txWithReduceMems, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream)
1113 : {
1114 0 : std::vector<WqeInfo> wqeInfoVec;
1115 0 : wqeInfoVec.reserve(WQE_RESERVE_LENGTH);
1116 0 : struct WrAuxInfo aux = {};
1117 0 : aux.dataType = RDMA_REDUCE_DATA_TYPE_TABLE[datatype];
1118 0 : aux.reduceType = RDMA_REDUCE_OP_TYPE_TABLE[redOp];
1119 0 : if (aux.dataType == static_cast<uint8_t>(RdmaReduceDataType::RDMA_REDUCE_DATA_INVALID)
1120 0 : || aux.reduceType == static_cast<uint8_t>(RdmaReduceOpType::RDMA_REDUCE_OP_INVALID)) {
1121 0 : HCCL_ERROR(
1122 : "unsupported data type [%s] or Reduce type [%s]", GetDataTypeEnumStr(datatype).c_str(),
1123 : GetReduceOpEnumStr(redOp).c_str());
1124 0 : return HCCL_E_INTERNAL;
1125 : }
1126 :
1127 0 : for (const TxMemoryInfo& txWithReduceMem : txWithReduceMems) {
1128 0 : CHK_RET(TxPayLoad(
1129 : txWithReduceMem.dstMemType, txWithReduceMem.dstOffset, txWithReduceMem.src, txWithReduceMem.len,
1130 : WqeType::WQE_TYPE_DATA_WITH_REDUCE, aux, wqeInfoVec));
1131 : }
1132 :
1133 0 : CHK_RET(TxSendDataAndNotify(wqeInfoVec, stream, GetUseOneDoorbellValue()));
1134 0 : return HCCL_SUCCESS;
1135 0 : }
1136 :
1137 0 : bool TransportIbverbs::IsSupportTransportWithReduce()
1138 : {
1139 0 : if (machinePara_.deviceType == DevType::DEV_TYPE_910B || machinePara_.deviceType == DevType::DEV_TYPE_910_93) {
1140 0 : return true;
1141 : } else {
1142 0 : return false;
1143 : }
1144 : }
1145 :
1146 : // 910A1 不支持write with notify;
1147 : // 910A2 由于PCIE through 和 write with notify 冲突，所以不支持write with Notify;
1148 0 : bool TransportIbverbs::IsSupportRdmaNotify() { return false; }
1149 :
1150 1 : bool TransportIbverbs::IsTemplateMode()
1151 : {
1152 1 : if (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE
1153 1 : || machinePara_.deviceType == DevType::DEV_TYPE_910B || machinePara_.deviceType == DevType::DEV_TYPE_910_93) {
1154 0 : return false;
1155 : } else {
1156 1 : return true;
1157 : }
1158 : }
1159 :
1160 0 : HcclResult TransportIbverbs::GetIndOpRemoteMemDetails(MemDetails** remoteMem, uint32_t* memNum, HcclMemType memType)
1161 : {
1162 0 : CHK_PRT_RET(remoteMem == nullptr, HCCL_ERROR("[%s] remoteMem is nullptr", __func__), HCCL_E_PARA);
1163 0 : CHK_PRT_RET(memNum == nullptr, HCCL_ERROR("[%s] memNum is nullptr", __func__), HCCL_E_PARA);
1164 :
1165 0 : *remoteMem = nullptr;
1166 0 : *memNum = 0;
1167 : uint32_t memCount;
1168 0 : if (memType == HcclMemType::HCCL_MEM_TYPE_DEVICE) {
1169 0 : memCount = remoteUserDeviceMemMsg_.size();
1170 0 : } else if (memType == HcclMemType::HCCL_MEM_TYPE_HOST) {
1171 0 : memCount = remoteUserHostMemMsg_.size();
1172 : } else {
1173 0 : HCCL_ERROR("[%s] not support memType[%d]", __func__, memType);
1174 0 : return HCCL_E_INTERNAL;
1175 : }
1176 0 : if (memCount == 0) {
1177 0 : HCCL_DEBUG("[%s] No remote memory regions available", __func__);
1178 0 : return HCCL_SUCCESS;
1179 : }
1180 : // 外部需要手动释放内存
1181 0 : MemDetails* remoteMemDetails = static_cast<MemDetails*>(malloc(memCount * sizeof(MemDetails)));
1182 0 : CHK_PTR_NULL(remoteMemDetails);
1183 0 : uint32_t index = 0;
1184 0 : if (memType == HcclMemType::HCCL_MEM_TYPE_DEVICE) {
1185 0 : for (const auto& msg : remoteUserDeviceMemMsg_) {
1186 0 : remoteMemDetails[index].addr = reinterpret_cast<u64>(msg.addr);
1187 0 : remoteMemDetails[index].size = msg.len;
1188 0 : remoteMemDetails[index].key = msg.lkey;
1189 0 : index++;
1190 : }
1191 0 : } else if (memType == HcclMemType::HCCL_MEM_TYPE_HOST) {
1192 0 : for (const auto& msg : remoteUserHostMemMsg_) {
1193 0 : remoteMemDetails[index].addr = reinterpret_cast<u64>(msg.addr);
1194 0 : remoteMemDetails[index].size = msg.len;
1195 0 : remoteMemDetails[index].key = msg.lkey;
1196 0 : index++;
1197 : }
1198 : }
1199 0 : *memNum = memCount;
1200 0 : *remoteMem = remoteMemDetails;
1201 :
1202 0 : HCCL_DEBUG("[%s] Successfully returned %u remote memory regions", __func__, index);
1203 0 : return HCCL_SUCCESS;
1204 : }
1205 :
1206 0 : HcclResult TransportIbverbs::GetIndOpRemoteMem(HcclMem** remoteMem, uint32_t* memNum)
1207 : {
1208 0 : CHK_PRT_RET(remoteMem == nullptr, HCCL_ERROR("[GetIndOpRemoteMem] remoteMem is nullptr"), HCCL_E_PARA);
1209 0 : CHK_PRT_RET(memNum == nullptr, HCCL_ERROR("[GetIndOpRemoteMem] memNum is nullptr"), HCCL_E_PARA);
1210 :
1211 0 : *remoteMem = nullptr;
1212 0 : *memNum = 0;
1213 :
1214 0 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
1215 :
1216 0 : if (!remoteMemsPtr_) {
1217 0 : uint32_t totalCount = remoteUserDeviceMemMsg_.size() + remoteUserHostMemMsg_.size();
1218 0 : if (totalCount == 0) {
1219 0 : HCCL_INFO("[GetIndOpRemoteMem] No remote memory regions available");
1220 0 : return HCCL_SUCCESS;
1221 : }
1222 0 : remoteMemsPtr_ = std::make_unique<HcclMem[]>(totalCount);
1223 0 : CHK_PTR_NULL(remoteMemsPtr_);
1224 0 : uint32_t index = 0;
1225 0 : for (const auto& msg : remoteUserDeviceMemMsg_) {
1226 0 : remoteMemsPtr_[index].type = HcclMemType::HCCL_MEM_TYPE_DEVICE;
1227 0 : remoteMemsPtr_[index].addr = msg.addr;
1228 0 : remoteMemsPtr_[index].size = msg.len;
1229 0 : index++;
1230 : }
1231 0 : for (const auto& msg : remoteUserHostMemMsg_) {
1232 0 : remoteMemsPtr_[index].type = HcclMemType::HCCL_MEM_TYPE_HOST;
1233 0 : remoteMemsPtr_[index].addr = msg.addr;
1234 0 : remoteMemsPtr_[index].size = msg.len;
1235 0 : index++;
1236 : }
1237 0 : remoteMemsNum_ = totalCount;
1238 : }
1239 :
1240 0 : *memNum = remoteMemsNum_;
1241 0 : *remoteMem = remoteMemsPtr_.get();
1242 :
1243 0 : return HCCL_SUCCESS;
1244 0 : }
1245 :
1246 0 : HcclResult TransportIbverbs::GetRemoteMem(UserMemType memType, void** remotePtr)
1247 : {
1248 0 : switch (memType) {
1249 0 : case UserMemType::INPUT_MEM:
1250 : case UserMemType::OUTPUT_MEM:
1251 0 : *remotePtr = remoteMemMsg_[static_cast<u32>(memType)].addr;
1252 0 : break;
1253 :
1254 0 : default:
1255 0 : HCCL_ERROR("[Get][RemoteMem]not support dst_mem_type=%d", memType);
1256 0 : return HCCL_E_NOT_SUPPORT;
1257 : }
1258 0 : return HCCL_SUCCESS;
1259 : }
1260 :
1261 0 : HcclResult TransportIbverbs::GetRemoteMemSize(UserMemType memType, u64& size)
1262 : {
1263 0 : switch (memType) {
1264 0 : case UserMemType::INPUT_MEM:
1265 : case UserMemType::OUTPUT_MEM:
1266 0 : size = remoteMemMsg_[static_cast<u32>(memType)].len;
1267 0 : break;
1268 :
1269 0 : default:
1270 0 : HCCL_ERROR("[Get][RemoteMem]not support dst_mem_type=%d", memType);
1271 0 : return HCCL_E_NOT_SUPPORT;
1272 : }
1273 0 : return HCCL_SUCCESS;
1274 : }
1275 :
1276 : HcclResult
1277 0 : TransportIbverbs::TxSendDataAndNotifyWithSingleQP(std::vector<WqeInfo>& wqeInfoVec, Stream& stream, bool useOneDoorbell)
1278 : {
1279 0 : if (IsSupportRdmaNotify() && wqeInfoVec.size() > 0) {
1280 : // 支持RDMA NOTIFY时, 修改wqeInfoVec中最后一个wqe, 使其附带Notify信息
1281 0 : u32 offset = static_cast<u32>(remoteDataNotifyMsg_.offset);
1282 0 : if (wqeInfoVec.back().wqeData.op == RA_WR_RDMA_REDUCE_WRITE) {
1283 0 : wqeInfoVec.back().wqeData.op = RA_WR_RDMA_REDUCE_WRITE_WITH_NOTIFY;
1284 : } else {
1285 0 : wqeInfoVec.back().wqeData.op = RA_WR_RDMA_WRITE_WITH_NOTIFY;
1286 : }
1287 0 : wqeInfoVec.back().wqeData.aux.notifyOffset = offset;
1288 : } else {
1289 : // 发送data notify同步信息
1290 0 : struct WrAuxInfo aux = {};
1291 0 : void* remoteNotifyaddr = remoteDataNotifyMsg_.addr;
1292 0 : CHK_RET(AddWqeList(
1293 : remoteNotifyaddr, notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_,
1294 : WqeType::WQE_TYPE_DATA_NOTIFY, aux, wqeInfoVec));
1295 : }
1296 :
1297 0 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
1298 0 : CHK_RET(RdmaSendAsync(wqeInfoVec, stream, useOneDoorbell));
1299 : } else {
1300 0 : CHK_RET(RdmaSendAsyncHostNIC(wqeInfoVec, stream));
1301 : }
1302 0 : return HCCL_SUCCESS;
1303 : }
1304 :
1305 0 : u32 TransportIbverbs::GetActualQpNum(u32 maxLength)
1306 : {
1307 0 : u32 actualMultiQpNum = 1;
1308 0 : const u32 KByteToByte = 1024; // 1024 多QP阈值单位是KB
1309 0 : if (maxLength / qpsPerConnection_ >= GetExternalInputMultiQpThreshold() * KByteToByte) {
1310 0 : actualMultiQpNum = qpsPerConnection_;
1311 : } else {
1312 0 : u32 quotient = maxLength / (GetExternalInputMultiQpThreshold() * KByteToByte);
1313 0 : u32 remainder = maxLength % (GetExternalInputMultiQpThreshold() * KByteToByte);
1314 0 : actualMultiQpNum = quotient + (remainder != 0 ? 1 : 0);
1315 : }
1316 :
1317 0 : return actualMultiQpNum;
1318 : }
1319 :
1320 0 : HcclResult TransportIbverbs::TxSendDataAndNotify(std::vector<WqeInfo>& wqeInfoVec, Stream& stream, bool useOneDoorbell)
1321 : {
1322 0 : u32 maxLength = 0;
1323 0 : for (u32 i = 0; i < wqeInfoVec.size(); i++) {
1324 0 : if (wqeInfoVec[i].wqeData.memList.len > maxLength) {
1325 0 : maxLength = wqeInfoVec[i].wqeData.memList.len;
1326 : }
1327 : }
1328 :
1329 0 : u32 actualMultiQpNum = GetActualQpNum(maxLength);
1330 :
1331 0 : HCCL_DEBUG(
1332 : "[TransportIbverbs][TxSendDataAndNotify] UseMultiQp[%d] MultiQpNum[%u] actualMultiQpNum[%u] maxLength[%u]",
1333 : UseMultiQp(), qpsPerConnection_, actualMultiQpNum, maxLength);
1334 0 : if (UseMultiQp() && actualMultiQpNum != 1 && actualMultiQpNum <= qpsPerConnection_ && maxLength != 0) {
1335 0 : CHK_RET(TxSendDataAndNotifyWithMultiQP(wqeInfoVec, actualMultiQpNum, stream, useOneDoorbell));
1336 : } else {
1337 0 : CHK_RET(TxSendDataAndNotifyWithSingleQP(wqeInfoVec, stream, useOneDoorbell));
1338 : }
1339 0 : return HCCL_SUCCESS;
1340 : }
1341 :
1342 0 : std::vector<u32> TransportIbverbs::RdmaLengthSplit(u32 length, u32 splitNum)
1343 : {
1344 : // step 1, 先计算有多少个128 Byte
1345 0 : u32 alignNum = length / RDMA_ADDR_ALIGNMENT;
1346 0 : u32 tailBytes = length % RDMA_ADDR_ALIGNMENT; // 尾块简单处理，放在最后一个切分出来的块后面
1347 : // step 2, 将这128 Byte再分成 splitNum 分，每一份有多少个 128Byte
1348 0 : u32 alignNumPerSplit = alignNum / splitNum;
1349 0 : u32 tailAlignNum = alignNum % splitNum; // 尾块简单处理，放在最后一个切分出来的块后面
1350 0 : std::vector<u32> vctSplittedLength(splitNum, 0);
1351 0 : for (u32 i = 0; i < splitNum; i++) {
1352 0 : u32 lengthTmp = alignNumPerSplit * RDMA_ADDR_ALIGNMENT;
1353 0 : vctSplittedLength[i] = lengthTmp;
1354 : }
1355 0 : vctSplittedLength[splitNum - 1] += tailAlignNum * RDMA_ADDR_ALIGNMENT + tailBytes;
1356 0 : return vctSplittedLength;
1357 : }
1358 :
1359 0 : HcclResult TransportIbverbs::TxSendDataAndNotifyWithMultiQP(
1360 : std::vector<WqeInfo>& wqeInfoVec, u32 actualMultiQpNum, Stream& stream, [[maybe_unused]] bool useOneDoorbell)
1361 : {
1362 : // vector<WqeInfo> 是一个vector的原因是 单个wqe只能发2GB数据，如果超过2GB，就拆分到多个WqeInfo中了
1363 : // 多QP下，对每个WqeInfo都进行多QP切分，然后在收发每一个QP的数据
1364 0 : std::vector<std::vector<WqeInfo>> multiQpWqeInfoVct(actualMultiQpNum, wqeInfoVec);
1365 0 : for (u32 i = 0; i < wqeInfoVec.size(); i++) {
1366 0 : WqeInfo tmpWqeInfo = wqeInfoVec[i];
1367 0 : u32 curLen = tmpWqeInfo.wqeData.memList.len;
1368 0 : std::vector<u32> splittedLen = RdmaLengthSplit(curLen, actualMultiQpNum);
1369 0 : uint64_t curSrcAddr = tmpWqeInfo.wqeData.memList.addr;
1370 0 : uint64_t curDstAddr = tmpWqeInfo.wqeData.dstAddr;
1371 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
1372 0 : multiQpWqeInfoVct[qpIndex][i].wqeData.memList.len = splittedLen[qpIndex];
1373 0 : multiQpWqeInfoVct[qpIndex][i].wqeData.memList.addr = curSrcAddr;
1374 0 : multiQpWqeInfoVct[qpIndex][i].wqeData.dstAddr = curDstAddr;
1375 0 : curSrcAddr += splittedLen[qpIndex];
1376 0 : curDstAddr += splittedLen[qpIndex];
1377 : }
1378 0 : }
1379 : // 给每个QP最后增加一个属于该QP的DataNotify
1380 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
1381 0 : struct WrAuxInfo aux = {};
1382 0 : void* remoteNotifyaddr = multiQpDataNotifyRemoteMemMsg_[qpIndex].addr;
1383 0 : CHK_RET(AddWqeList(
1384 : remoteNotifyaddr, notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_,
1385 : WqeType::WQE_TYPE_DATA_NOTIFY, aux, multiQpWqeInfoVct[qpIndex]));
1386 : }
1387 : // useOneDoorbell 配置成true。最后一个payload去按doorbell
1388 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
1389 0 : CHK_RET(
1390 : RdmaSendAsync(multiQpWqeInfoVct[qpIndex], stream, true, qpIndex)); // 多QP使用同一个stream异步doorbell触发
1391 : }
1392 0 : return HCCL_SUCCESS;
1393 0 : }
1394 :
1395 0 : HcclResult TransportIbverbs::TxAsync(std::vector<TxMemoryInfo>& txMems, Stream& stream)
1396 : {
1397 0 : std::vector<WqeInfo> wqeInfoVec;
1398 0 : wqeInfoVec.reserve(WQE_RESERVE_LENGTH);
1399 0 : struct WrAuxInfo aux = {};
1400 :
1401 0 : for (auto& mem : txMems) {
1402 0 : HCCL_DEBUG("TX src[%p] len[%llu] dstOffset[%llu]", mem.src, mem.len, mem.dstOffset);
1403 0 : CHK_PTR_NULL(mem.src);
1404 0 : CHK_RET(TxPayLoad(mem.dstMemType, mem.dstOffset, mem.src, mem.len, WqeType::WQE_TYPE_DATA, aux, wqeInfoVec));
1405 : }
1406 :
1407 0 : CHK_RET(TxSendDataAndNotify(wqeInfoVec, stream, GetUseOneDoorbellValue()));
1408 0 : return HCCL_SUCCESS;
1409 0 : }
1410 :
1411 1 : HcclResult TransportIbverbs::TxWqeList(
1412 : std::vector<WqeInfo>& wqeInfoVec, Stream& stream, std::vector<struct SendWrRsp>& opRspVec, u32 multiQpIndex)
1413 : {
1414 : (void)stream;
1415 1 : if (!IsTemplateMode()) {
1416 0 : currentQP_ = 0;
1417 : } else {
1418 2 : if (sqeCounter_ < (HCCP_SQ_TEMPLATE_CAPACITY + 1)
1419 1 : && (sqeCounter_ + wqeInfoVec.size()) >= (HCCP_SQ_TEMPLATE_CAPACITY + 1)) {
1420 0 : currentQP_++;
1421 0 : sqeCounter_ = wqeInfoVec.size();
1422 : } else {
1423 1 : sqeCounter_ += wqeInfoVec.size();
1424 : }
1425 : }
1426 1 : CHK_PRT_RET(
1427 : currentQP_ >= combineQpHandles_.size(),
1428 : HCCL_ERROR(
1429 : "[TransportIbverbs][TxWqeList]errNo[0x%016llx] In lbv "
1430 : "exp, qp idx[%u] is invalid.",
1431 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), currentQP_),
1432 : HCCL_E_INTERNAL);
1433 :
1434 0 : HCCL_DEBUG("rdma tx send wqes: ra qp sqe counter:%u, current qp idx:%u", sqeCounter_, currentQP_);
1435 :
1436 0 : std::vector<SendWrlistDataExt> wqelisDatatVec;
1437 0 : for (u32 index = 0; index < wqeInfoVec.size(); index++) {
1438 : // 使能atomic write场景下，reduce的下一个notify的opcode要设置为atomic write
1439 0 : u32& preWrOpcode = multiQpIndex == RDMA_INVALID_QP_INDEX ? combineQpHandles_[currentQP_].preWrOpcode :
1440 0 : multiCombineQpHandles_[multiQpIndex].preWrOpcode;
1441 0 : ModifyAtomicWriteAfterReduce(
1442 0 : preWrOpcode, wqeInfoVec[index].wqeType, wqeInfoVec[index].wqeData.op,
1443 0 : wqeInfoVec[index].wqeData.ext.immData);
1444 :
1445 0 : wqelisDatatVec.push_back(wqeInfoVec[index].wqeData);
1446 : }
1447 :
1448 0 : u32 totalWqeCount = wqelisDatatVec.size();
1449 0 : struct SendWrlistDataExt* wqelist = wqelisDatatVec.data();
1450 0 : struct SendWrRsp* opRsp = opRspVec.data();
1451 :
1452 : // HCCP会校验 zero byte messages 的内存地址是否已注册MR。对于 zero byte messages 不下发WR，将opRsp设置为特殊值。
1453 : // 下发rdmasend task时检查该特殊值，如果zero byte message则不下发rdmasend task。
1454 0 : bool batchSendWr = true;
1455 0 : for (u32 i = 0; i < totalWqeCount; i++) {
1456 0 : if (wqelisDatatVec[i].memList.len == 0) {
1457 0 : batchSendWr = false;
1458 0 : break;
1459 : }
1460 : }
1461 : QpHandle currentQp;
1462 0 : if (multiQpIndex == RDMA_INVALID_QP_INDEX) {
1463 0 : currentQp = combineQpHandles_[currentQP_].qpHandle;
1464 : } else {
1465 0 : currentQp = multiCombineQpHandles_[multiQpIndex].qpHandle;
1466 : }
1467 0 : if (batchSendWr) {
1468 0 : CHK_RET(SendWqeList(currentQp, totalWqeCount, wqelist, opRsp));
1469 : } else {
1470 0 : for (u32 i = 0; i < totalWqeCount; i++) {
1471 0 : if (wqelisDatatVec[i].memList.len > 0) {
1472 0 : CHK_RET(SendWqeList(currentQp, 1U, &wqelist[i], &opRsp[i]));
1473 : } else {
1474 0 : opRsp[i].wqeTmp.sqIndex = INVALID_UINT;
1475 0 : opRsp[i].wqeTmp.wqeIndex = INVALID_UINT;
1476 0 : opRsp[i].db.dbIndex = INVALID_UINT;
1477 0 : opRsp[i].db.dbInfo = INVALID_U64;
1478 : }
1479 : }
1480 : }
1481 :
1482 0 : return HCCL_SUCCESS;
1483 0 : }
1484 :
1485 : HcclResult
1486 0 : TransportIbverbs::SendWqeList(QpHandle qpHandle, u32 wqeNum, struct SendWrlistDataExt* wqelist, struct SendWrRsp* opRsp)
1487 : {
1488 0 : unsigned int completeNum = 0;
1489 0 : HcclResult ret = HrtRaSendWrlistExt(qpHandle, wqelist, opRsp, wqeNum, &completeNum);
1490 0 : CHK_PRT_RET(
1491 : ret != HCCL_SUCCESS,
1492 : HCCL_ERROR("[TransportIbverbs][TxWqeList]In ibv send wq list, HrtRaSendWrlist failed.ret[%d]", ret),
1493 : HCCL_E_NETWORK);
1494 0 : return HCCL_SUCCESS;
1495 : }
1496 :
1497 : HcclResult
1498 1 : TransportIbverbs::RdmaSendAsync(std::vector<WqeInfo>& wqeInfoVec, Stream& stream, bool useOneDoorbell, u32 multiQpIndex)
1499 : {
1500 : HcclResult ret;
1501 :
1502 1 : std::vector<struct SendWrRsp> opRspVec(wqeInfoVec.size());
1503 1 : CHK_RET(TxWqeList(wqeInfoVec, stream, opRspVec, multiQpIndex));
1504 :
1505 0 : std::vector<SendWrlistDataExt> wqelistVec;
1506 0 : for (u32 index = 0; index < wqeInfoVec.size(); index++) {
1507 0 : wqelistVec.push_back(wqeInfoVec[index].wqeData);
1508 : }
1509 :
1510 0 : struct SendWr wr = {};
1511 0 : wr.bufNum = 1;
1512 0 : wr.op = 0;
1513 0 : wr.sendFlag = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
1514 0 : fence_ = false;
1515 :
1516 0 : if (useOneDoorbell && !IsTemplateMode()) {
1517 : // 内存块不连续时，只敲最后一次doorbell
1518 0 : wr.bufList = &wqelistVec.back().memList;
1519 0 : wr.dstAddr = static_cast<u64>(wqelistVec.back().dstAddr);
1520 : // 只敲一次doorbell时，len为所有非连续内存块长度总和+notify(4Bytes)
1521 0 : wr.bufList[0].len = std::accumulate(wqelistVec.begin(), wqelistVec.end(), 0llu, [](u64 acc, auto wqelist) {
1522 0 : return acc + wqelist.memList.len;
1523 : });
1524 :
1525 0 : const u32 dbIndex = static_cast<u32>(opRspVec.back().db.dbIndex);
1526 0 : const u64 dbInfo = static_cast<u64>(opRspVec.back().db.dbInfo);
1527 0 : ret = dispatcher_->RdmaSend(dbIndex, dbInfo, wr, stream, machinePara_.remoteWorldRank, isCapture_);
1528 0 : CHK_PRT_RET(
1529 : ret != HCCL_SUCCESS,
1530 : HCCL_ERROR(
1531 : "[TransportIbverbs][RdmaSendAsync][useOneDoorbell]errNo[0x%016llx] In lbv exp op base mode, "
1532 : "rdma send failed. dbIndex[%u] dbInfo[%llu]",
1533 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo),
1534 : ret);
1535 0 : HCCL_INFO("[TransportIbverbs][RdmaSendAsync][useOneDoorbell] db_index[%u], db_info[%llu]", dbIndex, dbInfo);
1536 0 : return HCCL_SUCCESS;
1537 : }
1538 :
1539 0 : for (u32 i = 0; i < wqeInfoVec.size(); i++) {
1540 0 : wr.bufList = &wqelistVec[i].memList;
1541 0 : wr.dstAddr = static_cast<u64>(wqelistVec[i].dstAddr);
1542 :
1543 0 : if (!IsTemplateMode()) {
1544 0 : u32 dbIndex = static_cast<u32>(opRspVec[i].db.dbIndex);
1545 0 : u64 dbInfo = static_cast<u64>(opRspVec[i].db.dbInfo);
1546 :
1547 : // op base 模式下的发送接口
1548 0 : if (wqeInfoVec[i].wqeType == static_cast<u64>(WqeType::WQE_TYPE_DATA)) {
1549 0 : ret = dispatcher_->RdmaSend(dbIndex, dbInfo, wr, stream, machinePara_.remoteWorldRank, isCapture_);
1550 : } else {
1551 0 : ret = dispatcher_->RdmaSend(
1552 0 : dbIndex, dbInfo, wr, stream, machinePara_.remoteWorldRank, wqeInfoVec[i].wqeDataOffset, isCapture_);
1553 : }
1554 0 : CHK_PRT_RET(
1555 : ret != HCCL_SUCCESS,
1556 : HCCL_ERROR(
1557 : "[TransportIbverbs][RdmaSendAsync]errNo[0x%016llx] In lbv exp op base mode, "
1558 : "rdma send failed. dbIndex[%u] dbInfo[%llu] wqe type[%llu] offset[%llu]",
1559 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo, wqeInfoVec[i].wqeType, wqeInfoVec[i].wqeDataOffset),
1560 : ret);
1561 : } else { // offline mode
1562 : // 下沉模式
1563 0 : if (wqeInfoVec[i].wqeType == static_cast<u64>(WqeType::WQE_TYPE_DATA)) {
1564 0 : ret = dispatcher_->RdmaSend(
1565 0 : opRspVec[i].wqeTmp.sqIndex, opRspVec[i].wqeTmp.wqeIndex, wr, stream, machinePara_.remoteWorldRank);
1566 : } else {
1567 0 : ret = dispatcher_->RdmaSend(
1568 0 : opRspVec[i].wqeTmp.sqIndex, opRspVec[i].wqeTmp.wqeIndex, wr, stream, machinePara_.remoteWorldRank,
1569 0 : wqeInfoVec[i].wqeDataOffset);
1570 : }
1571 0 : CHK_PRT_RET(
1572 : ret != HCCL_SUCCESS,
1573 : HCCL_ERROR(
1574 : "[TransportIbverbs][RdmaSendAsync]errNo[0x%016llx] In lbv exp offline mode, "
1575 : "rdma send failed. sq_index[%u] wqe_index[%u], offset[%llu]",
1576 : HCCL_ERROR_CODE(ret), opRspVec[i].wqeTmp.sqIndex, opRspVec[i].wqeTmp.wqeIndex,
1577 : wqeInfoVec[i].wqeDataOffset),
1578 : ret);
1579 : }
1580 : }
1581 0 : return HCCL_SUCCESS;
1582 1 : }
1583 0 : HcclResult TransportIbverbs::RdmaSendAsyncHostNIC(std::vector<WqeInfo>& wqeInfoVec, Stream& stream)
1584 : {
1585 : HcclResult ret;
1586 :
1587 0 : std::vector<SendWrlistDataExt> wqelistVec;
1588 0 : for (u32 index = 0; index < wqeInfoVec.size(); index++) {
1589 0 : wqelistVec.push_back(wqeInfoVec[index].wqeData);
1590 : }
1591 0 : struct SendWrRsp opRsp = {};
1592 0 : struct SendWrlistDataExt wr = {};
1593 0 : for (u32 i = 0; i < wqeInfoVec.size(); i++) {
1594 0 : wr.memList.addr = wqelistVec[i].memList.addr;
1595 0 : wr.memList.len = wqelistVec[i].memList.len;
1596 0 : wr.dstAddr = static_cast<u64>(wqelistVec[i].dstAddr);
1597 0 : wr.op = 0;
1598 0 : wr.sendFlags = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
1599 0 : fence_ = false;
1600 :
1601 0 : if (wqeInfoVec[i].wqeType == static_cast<u64>(WqeType::WQE_TYPE_DATA)) {
1602 0 : ret = dispatcher_->HostNicRdmaSend(
1603 0 : combineQpHandles_[0].qpHandle, wr, opRsp, stream, machinePara_.remoteWorldRank);
1604 : } else {
1605 0 : ret = dispatcher_->HostNicRdmaSend(
1606 0 : combineQpHandles_[0].qpHandle, wr, opRsp, stream, machinePara_.remoteWorldRank,
1607 0 : wqeInfoVec[i].wqeDataOffset);
1608 : }
1609 0 : CHK_PRT_RET(
1610 : ret != HCCL_SUCCESS,
1611 : HCCL_ERROR(
1612 : "[TransportIbverbs][RdmaSendAsyncHostNIC]errNo[0x%016llx] In lbv exp offline "
1613 : "mode, rdma send failed. sq_index[%u] wqe_index[%u], offset[%llu]",
1614 : HCCL_ERROR_CODE(ret), opRsp.wqeTmp.sqIndex, opRsp.wqeTmp.wqeIndex, wqeInfoVec[i].wqeDataOffset),
1615 : ret);
1616 : }
1617 0 : return HCCL_SUCCESS;
1618 0 : }
1619 0 : HcclResult TransportIbverbs::RdmaSendAsync(
1620 : struct SendWr& wr, Stream& stream, WqeType wqeType, u64 notifyOffset, [[maybe_unused]] u32 notifyId)
1621 : {
1622 : HcclResult ret;
1623 0 : struct SendWrRsp opRsp = {};
1624 0 : if (!IsTemplateMode()) {
1625 0 : currentQP_ = 0;
1626 : } else {
1627 0 : if (sqeCounter_ == HCCP_SQ_TEMPLATE_CAPACITY) {
1628 0 : currentQP_++;
1629 0 : sqeCounter_ = 1;
1630 : } else {
1631 0 : sqeCounter_++;
1632 : }
1633 : }
1634 0 : CHK_PRT_RET(
1635 : currentQP_ >= combineQpHandles_.size(),
1636 : HCCL_ERROR(
1637 : "[TransportIbverbs][RdmaSendAsync]errNo[0x%016llx] In "
1638 : "lbv exp, qp idx[%u] is invalid.",
1639 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), currentQP_),
1640 : HCCL_E_INTERNAL);
1641 0 : HCCL_DEBUG("rdma send async: ra qp sqe counter:%u, current qp idx:%u.", sqeCounter_, currentQP_);
1642 :
1643 0 : CHK_RET(HrtRaSendWr(combineQpHandles_[currentQP_].qpHandle, &wr, &opRsp));
1644 :
1645 0 : if (!IsTemplateMode()) {
1646 0 : u32 dbIndex = static_cast<u32>(opRsp.db.dbIndex);
1647 0 : u64 dbInfo = static_cast<u64>(opRsp.db.dbInfo);
1648 0 : if (wqeType == WqeType::WQE_TYPE_DATA) {
1649 0 : ret = dispatcher_->RdmaSend(dbIndex, dbInfo, wr, stream, machinePara_.remoteWorldRank, isCapture_);
1650 0 : CHK_PRT_RET(
1651 : ret != HCCL_SUCCESS,
1652 : HCCL_ERROR(
1653 : "[TransportIbverbs][RdmaSendAsync]errNo[0x%016llx] In lbv exp op base mode, "
1654 : "rdma send failed. dbIndex[%u] dbInfo[%llu]",
1655 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo),
1656 : ret);
1657 : } else {
1658 0 : ret = dispatcher_->RdmaSend(
1659 0 : dbIndex, dbInfo, wr, stream, machinePara_.remoteWorldRank, notifyOffset, isCapture_);
1660 0 : CHK_PRT_RET(
1661 : ret != HCCL_SUCCESS,
1662 : HCCL_ERROR(
1663 : "[TransportIbverbs][RdmaSendAsync]errNo[0x%016llx] In lbv exp op base mode, "
1664 : "rdma send failed. dbIndex[%u] dbInfo[%llu], offset[%llu]",
1665 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo, notifyOffset),
1666 : ret);
1667 : }
1668 : } else { // offline mode
1669 0 : if (wqeType == WqeType::WQE_TYPE_DATA) {
1670 0 : ret = dispatcher_->RdmaSend(
1671 : opRsp.wqeTmp.sqIndex, opRsp.wqeTmp.wqeIndex, wr, stream, machinePara_.remoteWorldRank);
1672 0 : CHK_PRT_RET(
1673 : ret != HCCL_SUCCESS,
1674 : HCCL_ERROR(
1675 : "[TransportIbverbs][RdmaSendAsync]errNo[0x%016llx] In lbv exp offline mode, "
1676 : "rdma send failed. sq_index[%u] wqe_index[%u]",
1677 : HCCL_ERROR_CODE(ret), opRsp.wqeTmp.sqIndex, opRsp.wqeTmp.wqeIndex),
1678 : ret);
1679 : } else {
1680 0 : ret = dispatcher_->RdmaSend(
1681 : opRsp.wqeTmp.sqIndex, opRsp.wqeTmp.wqeIndex, wr, stream, machinePara_.remoteWorldRank, notifyOffset);
1682 0 : CHK_PRT_RET(
1683 : ret != HCCL_SUCCESS,
1684 : HCCL_ERROR(
1685 : "[TransportIbverbs][RdmaSendAsync]errNo[0x%016llx] In lbv exp offline mode, "
1686 : "rdma send failed. sq_index[%u] wqe_index[%u], offset[%llu]",
1687 : HCCL_ERROR_CODE(ret), opRsp.wqeTmp.sqIndex, opRsp.wqeTmp.wqeIndex, notifyOffset),
1688 : ret);
1689 : }
1690 : }
1691 0 : return HCCL_SUCCESS;
1692 : }
1693 : HcclResult
1694 0 : TransportIbverbs::RdmaSendAsyncHostNIC(struct SendWrlistDataExt& wr, Stream& stream, WqeType wqeType, u64 notifyOffset)
1695 : {
1696 : HcclResult ret;
1697 0 : struct SendWrRsp opRsp = {};
1698 0 : if (wqeType == WqeType::WQE_TYPE_DATA) {
1699 0 : ret = dispatcher_->HostNicRdmaSend(
1700 0 : combineQpHandles_[0].qpHandle, wr, opRsp, stream, machinePara_.remoteWorldRank);
1701 : } else {
1702 0 : ret = dispatcher_->HostNicRdmaSend(
1703 0 : combineQpHandles_[0].qpHandle, wr, opRsp, stream, machinePara_.remoteWorldRank, notifyOffset);
1704 : }
1705 0 : CHK_PRT_RET(
1706 : ret != HCCL_SUCCESS,
1707 : HCCL_ERROR(
1708 : "[TransportIbverbs][RdmaSendAsyncHostNIC]errNo[0x%016llx] In lbv exp offline mode, "
1709 : "rdma send failed. sq_index[%u] wqe_index[%u], offset[%llu]",
1710 : HCCL_ERROR_CODE(ret), opRsp.wqeTmp.sqIndex, opRsp.wqeTmp.wqeIndex, notifyOffset),
1711 : ret);
1712 :
1713 0 : return HCCL_SUCCESS;
1714 : }
1715 1 : HcclResult TransportIbverbs::GetWqeDataOffsetAndNotifyId(WqeType wqeType, u64& wqeDataOffset, u32& notifyId)
1716 : {
1717 1 : switch (wqeType) {
1718 1 : case WqeType::WQE_TYPE_DATA:
1719 : case WqeType::WQE_TYPE_DATA_WITH_NOTIFY:
1720 : case WqeType::WQE_TYPE_DATA_WITH_REDUCE:
1721 : case WqeType::WQE_TYPE_READ_DATA:
1722 1 : wqeDataOffset = 0;
1723 1 : notifyId = INVALID_UINT;
1724 1 : break;
1725 0 : case WqeType::WQE_TYPE_DATA_NOTIFY:
1726 0 : wqeDataOffset = remoteDataNotifyMsg_.offset;
1727 0 : notifyId = remoteDataNotifyMsg_.notifyId;
1728 0 : break;
1729 0 : case WqeType::WQE_TYPE_ACK_NOTIFY:
1730 0 : wqeDataOffset = remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].offset;
1731 0 : notifyId = remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].notifyId;
1732 0 : break;
1733 0 : case WqeType::WQE_TYPE_DATA_ACK_NOTIFY:
1734 0 : wqeDataOffset = remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].offset;
1735 0 : notifyId = remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].notifyId;
1736 0 : break;
1737 0 : default:
1738 0 : HCCL_ERROR("[Get][WqeDataOffset]error wqeType[%d]", wqeType);
1739 0 : return HCCL_E_INTERNAL;
1740 : }
1741 1 : return HCCL_SUCCESS;
1742 : }
1743 :
1744 : HcclResult
1745 0 : TransportIbverbs::TxSendWqe(void* dstMemPtr, const void* srcMemPtr, u64 srcMemSize, Stream& stream, WqeType wqeType)
1746 : {
1747 0 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !useAtomicWrite_) {
1748 0 : struct SgList list = {};
1749 0 : struct SendWr wr = {};
1750 : // 构造wr信息
1751 0 : list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
1752 0 : list.len = srcMemSize;
1753 :
1754 0 : wr.bufList = &list;
1755 0 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
1756 0 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(dstMemPtr));
1757 0 : wr.op = 0; /* RDMA_WRITE: 0 */
1758 0 : wr.sendFlag = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
1759 0 : fence_ = false;
1760 :
1761 : // 获取notify偏移地址，对于发送数据时，偏移地址为0
1762 0 : u32 notifyId = INVALID_UINT;
1763 0 : u64 wqeDataOffset = 0;
1764 0 : CHK_RET(GetWqeDataOffsetAndNotifyId(wqeType, wqeDataOffset, notifyId));
1765 :
1766 : // RDMA异步发送
1767 0 : CHK_RET(RdmaSendAsync(wr, stream, wqeType, wqeDataOffset, notifyId));
1768 0 : } else if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && useAtomicWrite_) {
1769 0 : struct WrAuxInfo aux = {};
1770 0 : std::vector<WqeInfo> wqeInfoVec;
1771 0 : CHK_RET(AddWqeList(dstMemPtr, srcMemPtr, srcMemSize, wqeType, aux, wqeInfoVec));
1772 0 : CHK_RET(RdmaSendAsync(wqeInfoVec, stream, false));
1773 0 : HCCL_DEBUG("TxSendWqe useAtomicWrite[%d]", useAtomicWrite_);
1774 0 : } else {
1775 0 : struct SendWrlistDataExt wr = {};
1776 : // 构造wr信息
1777 0 : wr.memList.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
1778 0 : wr.memList.len = srcMemSize;
1779 :
1780 0 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(dstMemPtr));
1781 0 : wr.op = 0; /* RDMA_WRITE: 0 */
1782 0 : wr.sendFlags = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
1783 0 : fence_ = false;
1784 :
1785 : // 获取notify偏移地址，对于发送数据时，偏移地址为0
1786 0 : u32 notifyId = INVALID_UINT;
1787 0 : u64 wqeDataOffset = 0;
1788 0 : CHK_RET(GetWqeDataOffsetAndNotifyId(wqeType, wqeDataOffset, notifyId));
1789 :
1790 : // RDMA异步发送
1791 0 : CHK_RET(RdmaSendAsyncHostNIC(wr, stream, wqeType, wqeDataOffset));
1792 : }
1793 0 : return HCCL_SUCCESS;
1794 : }
1795 :
1796 0 : HcclResult TransportIbverbs::TxSendNotifyWqe(MemMsg& memMsg, const void* srcMemPtr, u64 srcMemSize, Stream& stream)
1797 : {
1798 0 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && !useAtomicWrite_) {
1799 0 : struct SgList list = {};
1800 0 : struct SendWr wr = {};
1801 : // 构造wr信息
1802 0 : list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
1803 0 : list.len = srcMemSize;
1804 0 : wr.bufList = &list;
1805 0 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
1806 0 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(memMsg.addr));
1807 0 : wr.op = 0; /* RDMA_WRITE: 0 */
1808 0 : wr.sendFlag = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
1809 0 : fence_ = false;
1810 :
1811 : // RDMA异步发送
1812 0 : CHK_RET(RdmaSendAsync(wr, stream, WqeType::WQE_TYPE_ACK_NOTIFY, memMsg.offset, memMsg.notifyId));
1813 0 : } else if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE && useAtomicWrite_) {
1814 0 : struct WrAuxInfo aux = {};
1815 0 : std::vector<WqeInfo> wqeInfoVec;
1816 0 : CHK_RET(AddWqeList(memMsg.addr, srcMemPtr, srcMemSize, WqeType::WQE_TYPE_ACK_NOTIFY, aux, wqeInfoVec));
1817 0 : CHK_RET(RdmaSendAsync(wqeInfoVec, stream, false));
1818 0 : HCCL_DEBUG("TxSendNotifyWqe useAtomicWrite[%d]", useAtomicWrite_);
1819 0 : } else {
1820 0 : struct SendWrlistDataExt wr = {};
1821 : // 构造wr信息
1822 0 : wr.memList.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
1823 0 : wr.memList.len = srcMemSize;
1824 0 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(memMsg.addr));
1825 0 : wr.op = 0; /* RDMA_WRITE: 0 */
1826 0 : wr.sendFlags = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
1827 0 : fence_ = false;
1828 :
1829 : // RDMA异步发送
1830 0 : CHK_RET(RdmaSendAsyncHostNIC(wr, stream, WqeType::WQE_TYPE_ACK_NOTIFY, memMsg.offset));
1831 : }
1832 0 : return HCCL_SUCCESS;
1833 : }
1834 :
1835 : HcclResult
1836 0 : TransportIbverbs::RxAsync([[maybe_unused]] UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream)
1837 : {
1838 0 : u32 actualMultiQpNum = 1;
1839 0 : const u32 KByteToByte = 1024; // 1024 多QP阈值单位是KB
1840 0 : if (len / qpsPerConnection_ > GetExternalInputMultiQpThreshold() * KByteToByte) {
1841 0 : actualMultiQpNum = qpsPerConnection_;
1842 : } else {
1843 0 : u32 quotient = len / (GetExternalInputMultiQpThreshold() * KByteToByte);
1844 0 : u32 remainder = len % (GetExternalInputMultiQpThreshold() * KByteToByte);
1845 0 : actualMultiQpNum = quotient + (remainder != 0 ? 1 : 0);
1846 : }
1847 : // 等待TS把任务处理完成
1848 0 : HCCL_DEBUG(
1849 : "[TransportIbverbs][RxAsync] UseMultiQp[%d] actualMultiQpNum[%u], RX dst[%p] len[%llu] srcOffset[%llu]",
1850 : UseMultiQp(), actualMultiQpNum, dst, len, srcOffset);
1851 0 : if (UseMultiQp() && actualMultiQpNum != 1 && actualMultiQpNum <= qpsPerConnection_ && len != 0) {
1852 0 : for (u32 i = 0; i < actualMultiQpNum; i++) {
1853 0 : CHK_RET(LocalIpcNotify::Wait(
1854 : stream, dispatcher_, multiQpDataNotify_[i], INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME,
1855 : machinePara_.localUserrank, machinePara_.remoteWorldRank));
1856 : }
1857 : } else {
1858 0 : CHK_RET(LocalIpcNotify::Wait(
1859 : stream, dispatcher_, dataNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
1860 : machinePara_.remoteWorldRank));
1861 : }
1862 0 : return HCCL_SUCCESS;
1863 : }
1864 :
1865 0 : HcclResult TransportIbverbs::RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream& stream)
1866 : {
1867 0 : CHK_PRT_RET(rxMems.size() == 0, HCCL_ERROR("Invalid rxMem size[%u]", rxMems.size()), HCCL_E_PARA);
1868 0 : for (auto& mem : rxMems) {
1869 0 : HCCL_DEBUG("RX dst[%p] len[%llu] dstOffset[%llu]", mem.dst, mem.len, mem.srcOffset);
1870 : }
1871 0 : u32 maxLength = 0;
1872 0 : for (u32 i = 0; i < rxMems.size(); i++) {
1873 0 : if (rxMems[i].len > maxLength) {
1874 0 : maxLength = rxMems[i].len;
1875 : }
1876 : }
1877 :
1878 0 : CHK_RET(RxAsync(rxMems[0].srcMemType, rxMems[0].srcOffset, rxMems[0].dst, maxLength, stream));
1879 0 : return HCCL_SUCCESS;
1880 : }
1881 :
1882 0 : HcclResult TransportIbverbs::DataReceivedAck(Stream& stream)
1883 : {
1884 0 : CHK_RET(PostFinAck(stream));
1885 0 : CHK_RET(WaitFinAck(stream));
1886 :
1887 0 : return HCCL_SUCCESS;
1888 : }
1889 :
1890 0 : HcclResult TransportIbverbs::TxWaitDone([[maybe_unused]] Stream& stream) { return HCCL_SUCCESS; }
1891 :
1892 : /* 发送ack消息(同步模式) */
1893 0 : HcclResult TransportIbverbs::TxAck(Stream& stream)
1894 : {
1895 0 : CHK_RET(TxSendWqe(
1896 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr,
1897 : notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_, stream, WqeType::WQE_TYPE_ACK_NOTIFY));
1898 0 : return HCCL_SUCCESS;
1899 : }
1900 :
1901 : /* 接收ack消息(同步模式) */
1902 0 : HcclResult TransportIbverbs::RxAck(Stream& stream)
1903 : {
1904 0 : HcclResult ret = LocalIpcNotify::Wait(
1905 0 : stream, dispatcher_, ackNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
1906 : machinePara_.remoteWorldRank);
1907 0 : CHK_PRT_RET(
1908 : ret != HCCL_SUCCESS,
1909 : HCCL_ERROR(
1910 : "[TransportIbverbs][RxAck]errNo[0x%016llx] In lbv exp rx ack, signal wait failed. ", HCCL_ERROR_CODE(ret)),
1911 : ret);
1912 :
1913 0 : return HCCL_SUCCESS;
1914 : }
1915 :
1916 0 : HcclResult TransportIbverbs::TxDataSignal(Stream& stream)
1917 : {
1918 : // 发送data notify同步信息
1919 0 : void* remoteNotifyaddr = remoteDataNotifyMsg_.addr;
1920 0 : HcclResult ret = TxSendWqe(
1921 0 : remoteNotifyaddr, notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_, stream,
1922 : WqeType::WQE_TYPE_DATA_NOTIFY);
1923 0 : CHK_PRT_RET(
1924 : ret != HCCL_SUCCESS,
1925 : HCCL_ERROR(
1926 : "[TransportIbverbs][TxDataSignal]errNo[0x%016llx] In ibv tx data signal, send notify "
1927 : "wqe failed. dstMemPtr[%p], srcMemPtr[%p], srcMemSize[%llu Byte]",
1928 : HCCL_ERROR_CODE(ret), remoteNotifyaddr, notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_),
1929 : ret);
1930 : // 每发送一个data notify wqe, count 自增
1931 0 : return HCCL_SUCCESS;
1932 : }
1933 :
1934 0 : HcclResult TransportIbverbs::RxDataSignal(Stream& stream)
1935 : {
1936 : /* 等待send_ready_event事件 */
1937 0 : CHK_RET(LocalIpcNotify::Wait(
1938 : stream, dispatcher_, dataNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
1939 : machinePara_.remoteWorldRank));
1940 0 : return HCCL_SUCCESS;
1941 : }
1942 :
1943 0 : HcclResult TransportIbverbs::CreateNotifyVectorBuffer(
1944 : std::vector<std::shared_ptr<LocalIpcNotify>>& notifyVector, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
1945 : {
1946 0 : NotifyLoadType notifyLoadType
1947 0 : = machinePara_.isAicpuModeEn ? NotifyLoadType::DEVICE_NOTIFY : NotifyLoadType::HOST_NOTIFY;
1948 0 : for (u32 i = 0; i < notifyNum_; i++) {
1949 0 : std::shared_ptr<LocalIpcNotify> oneNotify;
1950 0 : CHK_RET(CreateNotifyBuffer(
1951 : oneNotify, MemType::MUILT_NOTIFY_MEM, exchangeDataPtr, exchangeDataBlankSize, notifyLoadType));
1952 0 : notifyVector.push_back(std::move(oneNotify));
1953 0 : }
1954 0 : return HCCL_SUCCESS;
1955 : }
1956 :
1957 0 : HcclResult TransportIbverbs::CreateNotifyBuffer(
1958 : std::shared_ptr<LocalIpcNotify>& localNotify, MemType notifyType, u8*& exchangeDataPtr, u64& exchangeDataBlankSize,
1959 : NotifyLoadType notifyLoadType)
1960 : {
1961 0 : u64 offset = 0;
1962 0 : u64 notifyBaseVa = 0; // notify寄存器虚拟地址
1963 0 : u64 notifyTotalSize = 0;
1964 0 : u32 notifyKey = 0;
1965 :
1966 0 : HcclRtNotify notify = nullptr;
1967 :
1968 : /* 获取notify寄存器虚拟基地址、大小, 物理地址回传值为空 */
1969 0 : struct MrInfoT mrInfo = {};
1970 0 : if (machinePara_.isAicpuModeEn
1971 0 : || (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE
1972 0 : && machinePara_.deviceType != localDeviceType)
1973 0 : || isHybridMode_) {
1974 0 : CHK_RET(HrtRaGetNotifyMrInfo(machinePara_.localDeviceId, nicRdmaHandle_, &mrInfo));
1975 0 : notifyBaseVa = reinterpret_cast<u64>(mrInfo.addr);
1976 0 : notifyTotalSize = mrInfo.size;
1977 0 : notifyKey = mrInfo.lkey;
1978 0 : } else {
1979 0 : u64 notifyBaseVaTmp = 0;
1980 0 : notifyBaseVaTmp = notifyBaseVa;
1981 0 : CHK_RET(HrtRaGetNotifyBaseAddr(nicRdmaHandle_, ¬ifyBaseVa, ¬ifyTotalSize, [this]() -> bool {
1982 : return this->GetStopFlag();
1983 : }));
1984 0 : CHK_PRT_RET(
1985 : ((notifyBaseVaTmp != 0) && (notifyBaseVaTmp != notifyBaseVa)),
1986 : HCCL_ERROR("[Create][NotifyBuffer]In lbv exp init, get base addr failed. notify base va has changed."),
1987 : HCCL_E_INTERNAL);
1988 : }
1989 :
1990 : /* 申请Notify Group ID */
1991 0 : RemoteRankInfo info(machinePara_.localDeviceId, machinePara_.remoteUserrank);
1992 0 : CHK_RET(SalGetBareTgid(&info.remotePid)); // 当前进程id
1993 :
1994 : // atomic write使能场景下，要求写入wr的notify地址是8byte对齐的
1995 0 : u32 offsetAlignSize = INVALID_UINT;
1996 0 : if (machinePara_.enableAtomicWrite) {
1997 : // 映射的MR基地址必定是8字节对齐的
1998 0 : CHK_PRT_RET(
1999 : notifyBaseVa % NOTIFY_VA_ALIGN_EIGHT != 0,
2000 : HCCL_ERROR("%s notifyBaseVa[0x%llx] not %u aligned", __func__, notifyBaseVa, NOTIFY_VA_ALIGN_EIGHT),
2001 : HCCL_E_INTERNAL);
2002 0 : offsetAlignSize = NOTIFY_VA_ALIGN_EIGHT;
2003 : }
2004 0 : CHK_RET(notifyPool_->Alloc(machinePara_.tag, info, localNotify, notifyLoadType, offsetAlignSize));
2005 : // 设置remote id
2006 0 : s64 recvId = 0xFFFFFFFF00000000 | (static_cast<s64>(info.remotePid) & 0xFFFFFFFF);
2007 0 : CHK_RET(localNotify->Grant(recvId));
2008 :
2009 : /* 获取notify虚拟地址 */
2010 0 : CHK_RET(localNotify->GetNotifyOffset(offset));
2011 :
2012 : // notify寄存器的虚拟地址与物理地址偏移相同，所以虚拟地址为虚拟基地址加偏移
2013 0 : u64 notifyVa = notifyBaseVa + offset;
2014 0 : CHK_PRT_RET(
2015 : machinePara_.enableAtomicWrite && (notifyVa % NOTIFY_VA_ALIGN_EIGHT != 0),
2016 : HCCL_ERROR(
2017 : "%s notifyVa[0x%llx] not %u aligned, notifyBaseVa[0x%llx], offset[0x%llx], enableAtomicWrite[%d]", __func__,
2018 : notifyVa, NOTIFY_VA_ALIGN_EIGHT, notifyBaseVa, offset, machinePara_.enableAtomicWrite),
2019 : HCCL_E_INTERNAL);
2020 :
2021 0 : HCCL_INFO(
2022 : "%s notifyBaseVa=0x%llx, notifyTotalSize=0x%x, offset=0x%llx, notifyVa=0x%llx machineType=%d, "
2023 : "notify=%p, notifyType=%d, notifyId=%u, offsetAlignSize[%u]",
2024 : __func__, notifyBaseVa, notifyTotalSize, offset, notifyVa, machinePara_.machineType, notify, notifyType,
2025 : localNotify->notifyId_, offsetAlignSize);
2026 :
2027 0 : if (notifyType != MULTI_QP_DATA_NOTIFY_MEM) {
2028 : /* notify地址注册为mr, 在roce驱动中注册 */
2029 0 : memMsg_[static_cast<u32>(notifyType)].mrRegFlag = 0; // mem注册给网卡标志位
2030 : // 本端notify地址交换给对端
2031 0 : memMsg_[static_cast<u32>(notifyType)].addr = reinterpret_cast<void*>(static_cast<uintptr_t>(notifyVa));
2032 0 : memMsg_[static_cast<u32>(notifyType)].len = notifySize_;
2033 0 : memMsg_[static_cast<u32>(notifyType)].memType = notifyType;
2034 0 : memMsg_[static_cast<u32>(notifyType)].offset = offset;
2035 0 : memMsg_[static_cast<u32>(notifyType)].lkey = mrInfo.lkey;
2036 0 : memMsg_[static_cast<u32>(notifyType)].lkey = notifyKey;
2037 0 : memMsg_[static_cast<u32>(notifyType)].notifyId = localNotify->notifyId_;
2038 :
2039 : /* 拼接要发送的数据 */
2040 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
2041 : exchangeDataPtr, exchangeDataBlankSize, reinterpret_cast<void*>(&memMsg_[static_cast<u32>(notifyType)]),
2042 : sizeof(MemMsg)));
2043 : } else {
2044 0 : MemMsg memMsg;
2045 0 : memMsg.mrRegFlag = 0;
2046 0 : memMsg.addr = reinterpret_cast<void*>(static_cast<uintptr_t>(notifyVa));
2047 0 : memMsg.len = notifySize_;
2048 0 : memMsg.memType = notifyType;
2049 0 : memMsg.offset = offset;
2050 0 : memMsg.notifyId = localNotify->notifyId_;
2051 0 : multiQpDataNotifyMemMsg_.push_back(std::move(memMsg));
2052 0 : CHK_SAFETY_FUNC_RET(
2053 : memcpy_s(exchangeDataPtr, exchangeDataBlankSize, reinterpret_cast<void*>(&memMsg), sizeof(MemMsg)));
2054 : }
2055 :
2056 0 : exchangeDataPtr += sizeof(MemMsg);
2057 0 : exchangeDataBlankSize -= sizeof(MemMsg);
2058 :
2059 0 : return HCCL_SUCCESS;
2060 : }
2061 :
2062 1 : HcclResult TransportIbverbs::RegUserMem(MemType memType, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
2063 : {
2064 1 : void* memPtr = nullptr;
2065 : u64 memSize;
2066 1 : switch (memType) {
2067 1 : case MemType::USER_INPUT_MEM: {
2068 1 : memPtr = machinePara_.inputMem.ptr();
2069 1 : memSize = machinePara_.inputMem.size();
2070 1 : break;
2071 : }
2072 :
2073 0 : case MemType::USER_OUTPUT_MEM: {
2074 0 : memPtr = machinePara_.outputMem.ptr();
2075 0 : memSize = machinePara_.outputMem.size();
2076 0 : break;
2077 : }
2078 :
2079 0 : default: {
2080 0 : HCCL_ERROR("[Reg][UserMem]not support dst_mem_type=%d", memType);
2081 0 : return HCCL_E_NOT_SUPPORT;
2082 : }
2083 : }
2084 1 : struct MrInfoT mrInfo = {};
2085 1 : mrInfo.addr = memPtr;
2086 1 : mrInfo.size = memSize;
2087 1 : mrInfo.access = access_;
2088 1 : if (mrInfo.size != 0) {
2089 1 : for (u32 i = 0; i < combineQpHandles_.size(); i++) {
2090 0 : CHK_RET(HrtRaMrReg(combineQpHandles_[i].qpHandle, &mrInfo));
2091 : }
2092 :
2093 1 : if (UseMultiQp()) {
2094 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
2095 0 : CHK_RET(HrtRaMrReg(multiCombineQpHandles_[i].qpHandle, &mrInfo));
2096 : }
2097 : }
2098 1 : memMsg_[static_cast<u32>(memType)].mrRegFlag = REG_VALID;
2099 : }
2100 :
2101 1 : memMsg_[static_cast<u32>(memType)].addr = memPtr;
2102 1 : memMsg_[static_cast<u32>(memType)].len = memSize;
2103 1 : memMsg_[static_cast<u32>(memType)].memType = memType;
2104 1 : memMsg_[static_cast<u32>(memType)].lkey = mrInfo.lkey;
2105 :
2106 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
2107 : exchangeDataPtr, exchangeDataBlankSize, reinterpret_cast<void*>(&memMsg_[static_cast<u32>(memType)]),
2108 : sizeof(MemMsg)));
2109 :
2110 1 : exchangeDataPtr += sizeof(MemMsg);
2111 1 : exchangeDataBlankSize -= sizeof(MemMsg);
2112 :
2113 1 : HCCL_DEBUG("memType=%d mem_ptr=%p mem_size=%llu Byte, key = %u", memType, memPtr, memSize, mrInfo.lkey);
2114 :
2115 1 : return HCCL_SUCCESS;
2116 : }
2117 :
2118 0 : HcclResult TransportIbverbs::RegCustomUserMemWithMsg(
2119 : void* addr, u64 size, MemMsg& memMsg, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
2120 : {
2121 0 : struct MrInfoT mrInfo = {};
2122 0 : mrInfo.addr = addr;
2123 0 : mrInfo.size = size;
2124 0 : mrInfo.access = access_;
2125 0 : for (u32 i = 0; i < combineQpHandles_.size(); i++) {
2126 0 : CHK_RET(HrtRaMrReg(combineQpHandles_[i].qpHandle, &mrInfo));
2127 : }
2128 :
2129 0 : if (UseMultiQp()) {
2130 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
2131 0 : CHK_RET(HrtRaMrReg(multiCombineQpHandles_[i].qpHandle, &mrInfo));
2132 : }
2133 : }
2134 :
2135 0 : memMsg.mrRegFlag = REG_VALID;
2136 0 : memMsg.addr = addr;
2137 0 : memMsg.len = size;
2138 0 : memMsg.lkey = mrInfo.lkey;
2139 :
2140 0 : CHK_SAFETY_FUNC_RET(
2141 : memcpy_s(exchangeDataPtr, exchangeDataBlankSize, reinterpret_cast<void*>(&memMsg), sizeof(MemMsg)));
2142 :
2143 0 : exchangeDataPtr += sizeof(MemMsg);
2144 0 : exchangeDataBlankSize -= sizeof(MemMsg);
2145 :
2146 0 : HCCL_DEBUG("mem_ptr=%p mem_size=%llu Byte, key = %u", addr, size, mrInfo.lkey);
2147 :
2148 0 : return HCCL_SUCCESS;
2149 : }
2150 :
2151 0 : HcclResult TransportIbverbs::RegCustomUserMem(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
2152 : {
2153 0 : u32 deviceMemNum = machinePara_.userDeviceMem.size();
2154 0 : s32 sRet = memcpy_s(exchangeDataPtr, sizeof(u32), reinterpret_cast<void*>(&deviceMemNum), sizeof(u32));
2155 0 : CHK_PRT_RET(
2156 : sRet != EOK,
2157 : HCCL_ERROR(
2158 : "[Set][LocalMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
2159 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
2160 : HCCL_E_MEMORY);
2161 0 : exchangeDataPtr += sizeof(u32);
2162 0 : exchangeDataBlankSize -= sizeof(u32);
2163 :
2164 0 : userDeviceMemMsg_.resize(deviceMemNum);
2165 0 : for (u32 i = 0; i < deviceMemNum; i++) {
2166 0 : RegCustomUserMemWithMsg(
2167 0 : machinePara_.userDeviceMem[i].ptr(), machinePara_.userDeviceMem[i].size(), userDeviceMemMsg_[i],
2168 : exchangeDataPtr, exchangeDataBlankSize);
2169 : }
2170 :
2171 0 : u32 hostMemNum = machinePara_.userHostMem.size();
2172 0 : sRet = memcpy_s(exchangeDataPtr, sizeof(u32), reinterpret_cast<void*>(&hostMemNum), sizeof(u32));
2173 0 : CHK_PRT_RET(
2174 : sRet != EOK,
2175 : HCCL_ERROR(
2176 : "[Set][LocalMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
2177 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
2178 : HCCL_E_MEMORY);
2179 0 : exchangeDataPtr += sizeof(u32);
2180 0 : exchangeDataBlankSize -= sizeof(u32);
2181 :
2182 0 : userHostMemMsg_.resize(hostMemNum);
2183 0 : for (u32 i = 0; i < hostMemNum; i++) {
2184 0 : RegCustomUserMemWithMsg(
2185 0 : machinePara_.userHostMem[i].ptr(), machinePara_.userHostMem[i].size(), userHostMemMsg_[i], exchangeDataPtr,
2186 : exchangeDataBlankSize);
2187 : }
2188 :
2189 0 : return HCCL_SUCCESS;
2190 : }
2191 :
2192 0 : HcclResult TransportIbverbs::GetMemInfo(UserMemType memType, void** dstMemPtr, u64* dstMemSize)
2193 : {
2194 0 : switch (memType) {
2195 0 : case UserMemType::INPUT_MEM: {
2196 0 : *dstMemPtr = remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].addr;
2197 0 : *dstMemSize = remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].len;
2198 0 : break;
2199 : }
2200 :
2201 0 : case UserMemType::OUTPUT_MEM: {
2202 0 : *dstMemPtr = remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].addr;
2203 0 : *dstMemSize = remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].len;
2204 0 : break;
2205 : }
2206 :
2207 0 : default: {
2208 0 : HCCL_ERROR("[Get][MemInfo]not support dst_mem_type=%d", memType);
2209 0 : return HCCL_E_NOT_SUPPORT;
2210 : }
2211 : }
2212 0 : return HCCL_SUCCESS;
2213 : }
2214 :
2215 0 : HcclResult TransportIbverbs::GetRemoteAddr(MemType memType, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
2216 : {
2217 0 : if (memType != MULTI_QP_DATA_NOTIFY_MEM) {
2218 0 : s32 sRet = memcpy_s(&remoteMemMsg_[static_cast<u32>(memType)], sizeof(MemMsg), exchangeDataPtr, sizeof(MemMsg));
2219 0 : CHK_PRT_RET(
2220 : sRet != EOK,
2221 : HCCL_ERROR(
2222 : "[Get][RemoteAddr]errNo[0x%016llx] In lbv exp get remote addr, "
2223 : "memcpy failed. errorno[%d], params:destMaxSize[%zu],count[%zu]",
2224 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(MemMsg), sizeof(MemMsg)),
2225 : HCCL_E_MEMORY);
2226 0 : CHK_PTR_NULL(remoteMemMsg_[static_cast<u32>(memType)].addr);
2227 : } else {
2228 0 : MemMsg memMsg;
2229 0 : s32 sRet = memcpy_s(&memMsg, sizeof(MemMsg), exchangeDataPtr, sizeof(MemMsg));
2230 0 : CHK_PRT_RET(
2231 : sRet != EOK,
2232 : HCCL_ERROR(
2233 : "[Get][RemoteAddr]errNo[0x%016llx] In lbv exp get remote addr, "
2234 : "memcpy failed. errorno[%d], params:destMaxSize[%zu],count[%zu]",
2235 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(MemMsg), sizeof(MemMsg)),
2236 : HCCL_E_MEMORY);
2237 0 : CHK_PTR_NULL(memMsg.addr);
2238 0 : multiQpDataNotifyRemoteMemMsg_.push_back(std::move(memMsg));
2239 : }
2240 :
2241 0 : exchangeDataPtr += sizeof(MemMsg);
2242 0 : exchangeDataBlankSize -= sizeof(MemMsg);
2243 0 : HCCL_INFO(
2244 : "GetRemoteAddr success: memType=%d, addr=%p len=%llu, notifyId=%u", static_cast<int32_t>(memType),
2245 : remoteMemMsg_[static_cast<u32>(memType)].addr, remoteMemMsg_[static_cast<u32>(memType)].len,
2246 : remoteMemMsg_[static_cast<u32>(memType)].notifyId);
2247 0 : return HCCL_SUCCESS;
2248 : }
2249 :
2250 0 : HcclResult TransportIbverbs::GetIndOpRemoteAddr(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
2251 : {
2252 0 : u32 remoteDmemNum = 0;
2253 0 : s32 sRet = memcpy_s(reinterpret_cast<void*>(&remoteDmemNum), sizeof(u32), exchangeDataPtr, sizeof(u32));
2254 0 : CHK_PRT_RET(
2255 : sRet != EOK,
2256 : HCCL_ERROR(
2257 : "[Get][RemoteMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
2258 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
2259 : HCCL_E_MEMORY);
2260 0 : exchangeDataPtr += sizeof(u32);
2261 0 : exchangeDataBlankSize -= sizeof(u32);
2262 :
2263 0 : remoteUserDeviceMemMsg_.resize(remoteDmemNum);
2264 0 : for (u32 i = 0; i < remoteDmemNum; i++) {
2265 0 : sRet = memcpy_s(&remoteUserDeviceMemMsg_[i], sizeof(MemMsg), exchangeDataPtr, sizeof(MemMsg));
2266 0 : CHK_PRT_RET(
2267 : sRet != EOK,
2268 : HCCL_ERROR(
2269 : "[Get][GetCustomRemoteAddr]errNo[0x%016llx] In lbv exp get remote addr, "
2270 : "memcpy failed. errorno[%d], params:destMaxSize[%zu],count[%zu]",
2271 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(MemMsg), sizeof(MemMsg)),
2272 : HCCL_E_MEMORY);
2273 0 : exchangeDataPtr += sizeof(MemMsg);
2274 0 : exchangeDataBlankSize -= sizeof(MemMsg);
2275 0 : CHK_PTR_NULL(remoteUserDeviceMemMsg_[i].addr);
2276 : }
2277 :
2278 0 : u32 remoteHmemNum = 0;
2279 0 : sRet = memcpy_s(reinterpret_cast<void*>(&remoteHmemNum), sizeof(u32), exchangeDataPtr, sizeof(u32));
2280 0 : CHK_PRT_RET(
2281 : sRet != EOK,
2282 : HCCL_ERROR(
2283 : "[Get][RemoteMem]errNo[0x%016llx] memory copy failed. errorno[%d], params:dstMaxSize[%zu],cnt[%zu]",
2284 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(u32), sizeof(u32)),
2285 : HCCL_E_MEMORY);
2286 0 : exchangeDataPtr += sizeof(u32);
2287 0 : exchangeDataBlankSize -= sizeof(u32);
2288 :
2289 0 : remoteUserHostMemMsg_.resize(remoteHmemNum);
2290 0 : for (u32 i = 0; i < remoteHmemNum; i++) {
2291 0 : sRet = memcpy_s(&remoteUserHostMemMsg_[i], sizeof(MemMsg), exchangeDataPtr, sizeof(MemMsg));
2292 0 : CHK_PRT_RET(
2293 : sRet != EOK,
2294 : HCCL_ERROR(
2295 : "[Get][GetCustomRemoteAddr]errNo[0x%016llx] In lbv exp get remote addr, "
2296 : "memcpy failed. errorno[%d], params:destMaxSize[%zu],count[%zu]",
2297 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(MemMsg), sizeof(MemMsg)),
2298 : HCCL_E_MEMORY);
2299 0 : exchangeDataPtr += sizeof(MemMsg);
2300 0 : exchangeDataBlankSize -= sizeof(MemMsg);
2301 0 : CHK_PTR_NULL(remoteUserHostMemMsg_[i].addr);
2302 : }
2303 :
2304 0 : return HCCL_SUCCESS;
2305 : }
2306 :
2307 0 : HcclResult TransportIbverbs::GetRemoteNotifyAddr(u8*& exchangeDataPtr, u64& exchangeDataBlankSize, MemMsg& memMsg)
2308 : {
2309 0 : s32 sRet = memcpy_s(&memMsg, sizeof(MemMsg), exchangeDataPtr, sizeof(MemMsg));
2310 0 : CHK_PRT_RET(
2311 : sRet != EOK,
2312 : HCCL_ERROR(
2313 : "[Get][GetRemoteNotifyAddr]errNo[0x%016llx] In lbv exp get remote addr, "
2314 : "memcpy failed. errorno[%d], params:destMaxSize[%zu],count[%zu]",
2315 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(MemMsg), sizeof(MemMsg)),
2316 : HCCL_E_MEMORY);
2317 :
2318 0 : exchangeDataPtr += sizeof(MemMsg);
2319 0 : exchangeDataBlankSize -= sizeof(MemMsg);
2320 0 : CHK_PTR_NULL(memMsg.addr);
2321 0 : HCCL_INFO("GetRemoteNotifyAddr success: addr=%p len=%llu", memMsg.addr, memMsg.len);
2322 0 : return HCCL_SUCCESS;
2323 : }
2324 :
2325 0 : HcclResult TransportIbverbs::CreateNotifyValueBuffer()
2326 : {
2327 0 : if (!machinePara_.userMemEnable && !machinePara_.drainEnable) {
2328 0 : HCCL_INFO("userMemEnable is false, no need to create notify value buffer");
2329 0 : return HCCL_SUCCESS;
2330 : }
2331 0 : std::unique_lock<std::mutex> lock(notifyValueMutex_[machinePara_.deviceLogicId]);
2332 0 : if (notifyValueMem_[machinePara_.deviceLogicId].ptr() == nullptr) {
2333 0 : u64 notifyVaule = 1; // notify值写1表示record
2334 0 : CHK_RET(DeviceMem::alloc(notifyValueMem_[machinePara_.deviceLogicId], notifyValueSize_));
2335 0 : HCCL_DEBUG(
2336 : "create notify value buffer[%p], size[%u]", notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_);
2337 :
2338 0 : CHK_RET(hrtMemSyncCopy(
2339 : notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifyValueMem_[machinePara_.deviceLogicId].size(),
2340 : ¬ifyVaule, notifySize_, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
2341 : }
2342 0 : lock.unlock();
2343 :
2344 0 : struct MrInfoT mrInfo = {};
2345 0 : mrInfo.addr = notifyValueMem_[machinePara_.deviceLogicId].ptr();
2346 0 : mrInfo.size = notifySize_;
2347 0 : mrInfo.access = access_;
2348 :
2349 0 : for (u32 i = 0; i < combineQpHandles_.size(); i++) {
2350 0 : CHK_RET(HrtRaMrReg(combineQpHandles_[i].qpHandle, &mrInfo));
2351 : }
2352 :
2353 0 : if (UseMultiQp()) {
2354 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
2355 0 : CHK_RET(HrtRaMrReg(multiCombineQpHandles_[i].qpHandle, &mrInfo));
2356 : }
2357 : }
2358 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].mrRegFlag = REG_VALID;
2359 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr = notifyValueMem_[machinePara_.deviceLogicId].ptr();
2360 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].len = notifySize_;
2361 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].memType = MemType::NOTIFY_SRC_MEM;
2362 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey = mrInfo.lkey;
2363 :
2364 0 : HCCL_DEBUG("notifyValueMem_=%p", notifyValueMem_[machinePara_.deviceLogicId].ptr());
2365 :
2366 0 : return HCCL_SUCCESS;
2367 0 : }
2368 34 : void TransportIbverbs::DestroySignal()
2369 : {
2370 34 : dataNotify_ = nullptr;
2371 :
2372 34 : ackNotify_ = nullptr;
2373 :
2374 34 : dataAckNotify_ = nullptr;
2375 :
2376 34 : multiQpDataNotify_.clear();
2377 34 : }
2378 :
2379 : /* 发送ack消息(同步模式) */
2380 0 : HcclResult TransportIbverbs::TxPrepare(Stream& stream)
2381 : {
2382 0 : HcclResult ret = LocalIpcNotify::Wait(
2383 0 : stream, dispatcher_, ackNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
2384 : machinePara_.remoteWorldRank);
2385 0 : CHK_PRT_RET(
2386 : ret != HCCL_SUCCESS,
2387 : HCCL_ERROR(
2388 : "[TransportIbverbs][TxPrepare]errNo[0x%016llx] In lbv exp rx ack, signal wait failed. ",
2389 : HCCL_ERROR_CODE(ret)),
2390 : ret);
2391 :
2392 0 : return HCCL_SUCCESS;
2393 : }
2394 :
2395 : /* 接收ack消息(同步模式) */
2396 0 : HcclResult TransportIbverbs::RxPrepare(Stream& stream)
2397 : {
2398 0 : CHK_RET(TxSendWqe(
2399 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr,
2400 : notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_, stream, WqeType::WQE_TYPE_ACK_NOTIFY));
2401 0 : return HCCL_SUCCESS;
2402 : }
2403 :
2404 0 : HcclResult TransportIbverbs::TxData(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
2405 : {
2406 0 : std::vector<WqeInfo> wqeInfoVec;
2407 0 : struct WrAuxInfo aux = {};
2408 0 : HCCL_DEBUG("TX src[%p] len[%llu] dstOffset[%llu]", src, len, dstOffset);
2409 :
2410 0 : if (src != nullptr) {
2411 0 : CHK_RET(TxPayLoad(dstMemType, dstOffset, src, len, WqeType::WQE_TYPE_DATA, aux, wqeInfoVec));
2412 : }
2413 :
2414 0 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
2415 0 : CHK_RET(RdmaSendAsync(wqeInfoVec, stream, false));
2416 : } else {
2417 0 : CHK_RET(RdmaSendAsyncHostNIC(wqeInfoVec, stream));
2418 : }
2419 0 : return HCCL_SUCCESS;
2420 0 : }
2421 :
2422 0 : HcclResult TransportIbverbs::RxData(
2423 : [[maybe_unused]] UserMemType srcMemType, [[maybe_unused]] u64 srcOffset, [[maybe_unused]] void* dst,
2424 : [[maybe_unused]] u64 len, [[maybe_unused]] Stream& stream)
2425 : {
2426 0 : return HCCL_SUCCESS;
2427 : }
2428 :
2429 0 : HcclResult TransportIbverbs::TxDone(Stream& stream)
2430 : {
2431 : // 发送数据接收确认notify
2432 0 : CHK_RET(TxSendWqe(
2433 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].addr,
2434 : notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_, stream, WqeType::WQE_TYPE_DATA_NOTIFY));
2435 : // 接收数据接收确认notify
2436 0 : CHK_RET(LocalIpcNotify::Wait(
2437 : stream, dispatcher_, dataAckNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
2438 : machinePara_.remoteWorldRank));
2439 0 : return HCCL_SUCCESS;
2440 : }
2441 :
2442 0 : HcclResult TransportIbverbs::RxDone(Stream& stream)
2443 : {
2444 : // 接收数据接收确认notify
2445 0 : CHK_RET(LocalIpcNotify::Wait(
2446 : stream, dispatcher_, dataNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
2447 : machinePara_.remoteWorldRank));
2448 :
2449 : // 发送数据接收确认notify
2450 0 : CHK_RET(TxSendWqe(
2451 : remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].addr,
2452 : notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_, stream, WqeType::WQE_TYPE_DATA_ACK_NOTIFY));
2453 0 : return HCCL_SUCCESS;
2454 : }
2455 :
2456 0 : HcclResult TransportIbverbs::PostReady(Stream& stream)
2457 : {
2458 0 : CHK_RET(TxSendWqe(
2459 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr,
2460 : notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_, stream, WqeType::WQE_TYPE_ACK_NOTIFY));
2461 0 : return HCCL_SUCCESS;
2462 : }
2463 :
2464 0 : HcclResult TransportIbverbs::WaitReady(Stream& stream)
2465 : {
2466 0 : CHK_RET(LocalIpcNotify::Wait(
2467 : stream, dispatcher_, ackNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
2468 : machinePara_.remoteWorldRank));
2469 0 : return HCCL_SUCCESS;
2470 : }
2471 :
2472 0 : HcclResult TransportIbverbs::PostFin(Stream& stream)
2473 : {
2474 : // 发送data notify同步信息
2475 0 : void* remoteNotifyaddr = remoteDataNotifyMsg_.addr;
2476 0 : HcclResult ret = TxSendWqe(
2477 0 : remoteNotifyaddr, notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_, stream,
2478 : WqeType::WQE_TYPE_DATA_NOTIFY);
2479 0 : CHK_PRT_RET(
2480 : ret != HCCL_SUCCESS,
2481 : HCCL_ERROR(
2482 : "[TransportIbverbs][PostFin]errNo[0x%016llx] In ibv tx data signal, send notify "
2483 : "wqe failed. dstMemPtr[%p], srcMemPtr[%p], srcMemSize[%llu]",
2484 : HCCL_ERROR_CODE(ret), remoteNotifyaddr, notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_),
2485 : ret);
2486 : // 每发送一个data notify wqe, count 自增
2487 0 : return HCCL_SUCCESS;
2488 : }
2489 :
2490 0 : HcclResult TransportIbverbs::WaitFin(Stream& stream)
2491 : {
2492 0 : CHK_RET(LocalIpcNotify::Wait(
2493 : stream, dispatcher_, dataNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
2494 : machinePara_.remoteWorldRank));
2495 0 : return HCCL_SUCCESS;
2496 : }
2497 :
2498 0 : HcclResult TransportIbverbs::PostFinAck(Stream& stream)
2499 : {
2500 0 : CHK_RET(TxSendWqe(
2501 : remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].addr,
2502 : notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_, stream, WqeType::WQE_TYPE_DATA_ACK_NOTIFY));
2503 0 : return HCCL_SUCCESS;
2504 : }
2505 :
2506 0 : HcclResult TransportIbverbs::WaitFinAck(Stream& stream)
2507 : {
2508 0 : CHK_RET(LocalIpcNotify::Wait(
2509 : stream, dispatcher_, dataAckNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
2510 : machinePara_.remoteWorldRank));
2511 0 : return HCCL_SUCCESS;
2512 : }
2513 :
2514 0 : HcclResult TransportIbverbs::Post(u32 notifyIdx, Stream& stream)
2515 : {
2516 : // 校验notifyIdx有效性
2517 0 : bool bRet = (notifyIdx >= notifyNum_);
2518 0 : CHK_PRT_RET(
2519 : bRet,
2520 : HCCL_ERROR(
2521 : "[TransportIbverbs][Post]notifyNum[%u], notifyIdx[%u] out of range[0, %u]", notifyNum_, notifyIdx,
2522 : notifyNum_ - 1),
2523 : HCCL_E_INTERNAL);
2524 :
2525 : // 每个QP发送一个指定idx的notify
2526 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
2527 0 : CHK_RET(TxSendNotifyWqe(
2528 : userMultiQpRemoteNotifyMsg_[i][notifyIdx], notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize_,
2529 : stream));
2530 : }
2531 0 : return HCCL_SUCCESS;
2532 : }
2533 :
2534 0 : HcclResult TransportIbverbs::Wait(u32 notifyIdx, Stream& stream, const u32 timeOut)
2535 : {
2536 : // 校验notifyIdx有效性
2537 0 : bool bRet = (notifyIdx >= notifyNum_);
2538 0 : CHK_PRT_RET(
2539 : bRet,
2540 : HCCL_ERROR(
2541 : "[TransportIbverbs][Wait]notifyNum[%u], notifyIdx[%u] out of range[0, %u]", notifyNum_, notifyIdx,
2542 : notifyNum_ - 1),
2543 : HCCL_E_INTERNAL);
2544 :
2545 : // 每个qp接收一个指定idx的notify
2546 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
2547 0 : CHK_RET(LocalIpcNotify::Wait(
2548 : stream, dispatcher_, userMultiQpLocalNotify_[i][notifyIdx], INVALID_VALUE_STAGE, timeOut,
2549 : machinePara_.localUserrank, machinePara_.remoteWorldRank));
2550 : }
2551 0 : return HCCL_SUCCESS;
2552 : }
2553 :
2554 0 : HcclResult TransportIbverbs::GetLocalRdmaNotify(std::vector<HcclSignalInfo>& rdmaNotify)
2555 : {
2556 : HcclSignalInfo signalInfo;
2557 0 : CHK_SMART_PTR_NULL(ackNotify_);
2558 0 : CHK_RET(ackNotify_->GetNotifyData(signalInfo));
2559 0 : rdmaNotify.push_back(signalInfo);
2560 0 : CHK_SMART_PTR_NULL(ackNotify_);
2561 0 : CHK_RET(dataNotify_->GetNotifyData(signalInfo));
2562 0 : rdmaNotify.push_back(signalInfo);
2563 0 : CHK_SMART_PTR_NULL(ackNotify_);
2564 0 : CHK_RET(dataAckNotify_->GetNotifyData(signalInfo));
2565 0 : rdmaNotify.push_back(signalInfo);
2566 : // 提取新增的notify资源
2567 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
2568 0 : for (u32 j = 0; j < notifyNum_; j++) {
2569 0 : CHK_SMART_PTR_NULL(userMultiQpLocalNotify_[i][j]);
2570 0 : CHK_RET(userMultiQpLocalNotify_[i][j]->GetNotifyData(signalInfo));
2571 0 : rdmaNotify.push_back(signalInfo);
2572 : }
2573 0 : if (qpsPerConnection_ > 1) {
2574 0 : CHK_SMART_PTR_NULL(multiQpDataNotify_[i]);
2575 0 : CHK_RET(multiQpDataNotify_[i]->GetNotifyData(signalInfo));
2576 0 : rdmaNotify.push_back(signalInfo);
2577 : }
2578 0 : HCCL_DEBUG(
2579 : "[TransportIbverbs][GetLocalRdmaNotify] resId[%llu] addr[%llu]", rdmaNotify.back().resId,
2580 : rdmaNotify.back().addr);
2581 : }
2582 0 : return HCCL_SUCCESS;
2583 : }
2584 :
2585 0 : HcclResult TransportIbverbs::GetRemoteRdmaNotifyAddrKey(std::vector<AddrKey>& rdmaNotify)
2586 : {
2587 0 : if (remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr == nullptr) {
2588 0 : HCCL_ERROR("[TransportIbverbs][GetRemoteRdmaNotifyAddrKey] ackNotify is null!");
2589 0 : return HCCL_E_PTR;
2590 0 : } else if (remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].addr == nullptr) {
2591 0 : HCCL_ERROR("[TransportIbverbs][GetRemoteRdmaNotifyAddrKey] dataNotify is null!");
2592 0 : return HCCL_E_PTR;
2593 0 : } else if (remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].addr == nullptr) {
2594 0 : HCCL_ERROR("[TransportIbverbs][GetRemoteRdmaNotifyAddrKey] dataAckNotify is null!");
2595 0 : return HCCL_E_PTR;
2596 : }
2597 0 : AddrKey notifyDetails;
2598 0 : notifyDetails.addr = reinterpret_cast<u64>(remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr);
2599 0 : notifyDetails.key = reinterpret_cast<u32>(remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].lkey);
2600 0 : notifyDetails.notifyId = remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].notifyId;
2601 0 : rdmaNotify.push_back(notifyDetails);
2602 0 : notifyDetails.addr = reinterpret_cast<u64>(remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].addr);
2603 0 : notifyDetails.key = reinterpret_cast<u32>(remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].lkey);
2604 0 : notifyDetails.notifyId = remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].notifyId;
2605 0 : rdmaNotify.push_back(notifyDetails);
2606 0 : notifyDetails.addr = reinterpret_cast<u64>(remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].addr);
2607 0 : notifyDetails.key = reinterpret_cast<u32>(remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].lkey);
2608 0 : notifyDetails.notifyId = remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].notifyId;
2609 0 : rdmaNotify.push_back(notifyDetails);
2610 :
2611 : // 获取新增的多notify资源
2612 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
2613 0 : for (u32 j = 0; j < notifyNum_; j++) {
2614 0 : notifyDetails.addr = reinterpret_cast<u64>(userMultiQpRemoteNotifyMsg_[i][j].addr);
2615 0 : notifyDetails.key = reinterpret_cast<u32>(userMultiQpRemoteNotifyMsg_[i][j].lkey);
2616 0 : notifyDetails.notifyId = userMultiQpRemoteNotifyMsg_[i][j].notifyId;
2617 0 : rdmaNotify.push_back(notifyDetails);
2618 : }
2619 0 : if (qpsPerConnection_ > 1) {
2620 0 : notifyDetails.addr = reinterpret_cast<u64>(multiQpDataNotifyRemoteMemMsg_[i].addr);
2621 0 : notifyDetails.key = reinterpret_cast<u32>(multiQpDataNotifyRemoteMemMsg_[i].lkey);
2622 0 : notifyDetails.notifyId = multiQpDataNotifyRemoteMemMsg_[i].notifyId;
2623 0 : rdmaNotify.push_back(notifyDetails);
2624 : }
2625 0 : HCCL_DEBUG(
2626 : "[TransportIbverbs][GetRemoteRdmaNotifyAddrKey]remote addr[0x%llx], key[%lu], notifyId[%u]",
2627 : rdmaNotify.back().addr, rdmaNotify.back().key, rdmaNotify.back().notifyId);
2628 : }
2629 0 : return HCCL_SUCCESS;
2630 : }
2631 :
2632 0 : HcclResult TransportIbverbs::GetLocalNotifyValueAddrKey(std::vector<AddrKey>& notifyValue)
2633 : {
2634 0 : if (memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr == nullptr) {
2635 0 : HCCL_ERROR("[TransportIbverbs][GetLocalNotifyValueAddrKey] notifyValue is null!");
2636 0 : return HCCL_E_PTR;
2637 : }
2638 :
2639 0 : AddrKey notifyDetails;
2640 0 : notifyDetails.addr = reinterpret_cast<u64>(memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr);
2641 0 : notifyDetails.key = reinterpret_cast<u32>(memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey);
2642 0 : notifyValue.push_back(notifyDetails);
2643 0 : return HCCL_SUCCESS;
2644 : }
2645 :
2646 0 : HcclResult TransportIbverbs::GetRemoteMemKey(UserMemType memType, uint32_t* remoteMemKey)
2647 : {
2648 0 : switch (memType) {
2649 0 : case UserMemType::INPUT_MEM:
2650 : case UserMemType::OUTPUT_MEM:
2651 0 : *remoteMemKey = remoteMemMsg_[static_cast<u32>(memType)].lkey;
2652 0 : break;
2653 :
2654 0 : default:
2655 0 : HCCL_ERROR("[Get][RemoteMemKey]not support dst_mem_type=%d", memType);
2656 0 : return HCCL_E_NOT_SUPPORT;
2657 : }
2658 0 : return HCCL_SUCCESS;
2659 : }
2660 :
2661 0 : HcclResult TransportIbverbs::GetLocalMemDetails(UserMemType memType, MemDetails& memDetails)
2662 : {
2663 0 : switch (memType) {
2664 0 : case UserMemType::INPUT_MEM:
2665 : case UserMemType::OUTPUT_MEM:
2666 0 : memDetails.addr = reinterpret_cast<u64>(memMsg_[static_cast<u32>(memType)].addr);
2667 0 : memDetails.size = memMsg_[static_cast<u32>(memType)].len;
2668 0 : memDetails.key = memMsg_[static_cast<u32>(memType)].lkey;
2669 0 : break;
2670 :
2671 0 : default:
2672 0 : HCCL_ERROR("[Get][LocalMemDetails]not support dst_mem_type=%d", memType);
2673 0 : return HCCL_E_NOT_SUPPORT;
2674 : }
2675 0 : return HCCL_SUCCESS;
2676 : }
2677 :
2678 0 : HcclResult TransportIbverbs::GetAiQpInfo(std::vector<HcclQpInfoV2>& aiQpInfo)
2679 : {
2680 0 : aiQpInfo.resize(combineAiQpInfos_.size() + 1);
2681 :
2682 0 : aiQpInfo[0].qpPtr = combineAiQpInfo_.aiQpInfo.aiQpAddr;
2683 0 : aiQpInfo[0].sqIndex = combineAiQpInfo_.aiQpInfo.sqIndex;
2684 0 : aiQpInfo[0].dbIndex = combineAiQpInfo_.aiQpInfo.dbIndex;
2685 0 : HCCL_DEBUG(
2686 : "[TransportIbverbs][GetAiQpInfo] i[0] qpPtr[%llu] sqIndex[%u] dbIndex[%u]", aiQpInfo[0].qpPtr,
2687 : aiQpInfo[0].sqIndex, aiQpInfo[0].dbIndex);
2688 0 : for (u32 i = 1, j = 0; i < aiQpInfo.size(); i++, j++) {
2689 0 : aiQpInfo[i].qpPtr = combineAiQpInfos_[j].aiQpInfo.aiQpAddr;
2690 0 : aiQpInfo[i].sqIndex = combineAiQpInfos_[j].aiQpInfo.sqIndex;
2691 0 : aiQpInfo[i].dbIndex = combineAiQpInfos_[j].aiQpInfo.dbIndex;
2692 0 : HCCL_DEBUG(
2693 : "[TransportIbverbs][GetAiQpInfo] i[%u] qpPtr[%llu] sqIndex[%u] dbIndex[%u]", i, aiQpInfo[i].qpPtr,
2694 : aiQpInfo[i].sqIndex, aiQpInfo[i].dbIndex);
2695 : }
2696 0 : return HCCL_SUCCESS;
2697 : }
2698 :
2699 0 : HcclResult TransportIbverbs::GetAiRMAQueueInfo(std::vector<HcclAiRMAQueueInfo>& aiRMAQueueInfo)
2700 : {
2701 0 : bool isSupport = false;
2702 0 : CHK_RET(IsSupportAIVNormalQP(machinePara_.localDeviceId, isSupport));
2703 0 : CHK_PRT_RET(
2704 : isSupport == false,
2705 : HCCL_ERROR(
2706 : "[IsSupportCQCoverNormalQP]"
2707 : "devicePhyId[%u] not support",
2708 : machinePara_.localDeviceId),
2709 : HCCL_E_NOT_SUPPORT);
2710 :
2711 0 : u32 sl = GetExternalInputRdmaServerLevel();
2712 0 : if (machinePara_.sl != HCCL_COMM_SERVICE_LEVEL_CONFIG_NOT_SET) {
2713 0 : sl = machinePara_.sl;
2714 : }
2715 0 : HCCL_INFO(
2716 : "[TransportIbverbs][GetAiRMAQueueInfo] localUserRank[%u], remoteUserrank[%u], sl[%u]",
2717 : machinePara_.localUserrank, machinePara_.remoteUserrank, sl);
2718 0 : aiRMAQueueInfo.resize(combineAiQpInfos_.size() + 1);
2719 0 : CopyAiWQInfo(aiRMAQueueInfo[0].sq, combineAiQpInfo_.aiQpInfo.dataPlaneInfo.sq, DBMode::HW_DB, sl);
2720 0 : CopyAiWQInfo(aiRMAQueueInfo[0].rq, combineAiQpInfo_.aiQpInfo.dataPlaneInfo.rq, DBMode::SW_DB, sl);
2721 0 : CopyAiCQInfo(aiRMAQueueInfo[0].scq, combineAiQpInfo_.aiQpInfo.dataPlaneInfo.scq, DBMode::SW_DB);
2722 0 : CopyAiCQInfo(aiRMAQueueInfo[0].rcq, combineAiQpInfo_.aiQpInfo.dataPlaneInfo.rcq, DBMode::SW_DB);
2723 :
2724 : // 预留多QP的能力; 当前主要是单QP场景
2725 0 : for (u32 i = 1, j = 0; i < aiRMAQueueInfo.size(); i++, j++) {
2726 0 : CopyAiWQInfo(aiRMAQueueInfo[i].sq, combineAiQpInfos_[j].aiQpInfo.dataPlaneInfo.sq, DBMode::HW_DB, sl);
2727 0 : CopyAiWQInfo(aiRMAQueueInfo[i].rq, combineAiQpInfos_[j].aiQpInfo.dataPlaneInfo.rq, DBMode::SW_DB, sl);
2728 0 : CopyAiCQInfo(aiRMAQueueInfo[i].scq, combineAiQpInfos_[j].aiQpInfo.dataPlaneInfo.scq, DBMode::SW_DB);
2729 0 : CopyAiCQInfo(aiRMAQueueInfo[i].rcq, combineAiQpInfos_[j].aiQpInfo.dataPlaneInfo.rcq, DBMode::SW_DB);
2730 : }
2731 0 : return HCCL_SUCCESS;
2732 : }
2733 :
2734 0 : HcclResult TransportIbverbs::WriteCommon(
2735 : const void* remoteAddr, const void* localAddr, u64 length, Stream& stream, WqeType wqeType, struct WrAuxInfo& aux)
2736 : {
2737 : // 单qp & 多qp
2738 0 : std::vector<WqeInfo> wqeInfoVec;
2739 0 : wqeInfoVec.reserve(WQE_RESERVE_LENGTH);
2740 0 : HCCL_DEBUG("write localAddr[%p] remoteAddr[%p] len[%llu] remoteOffset[%llu]", localAddr, remoteAddr, length);
2741 :
2742 0 : if (localAddr != nullptr) {
2743 : // 为保证单算子下不同数据量下子图的结构相同，zero byte message 时也需要下发task
2744 0 : u32 txSendDataTimes = (length == 0) ? 1 : (length + RDMA_SEND_MAX_SIZE - 1) / RDMA_SEND_MAX_SIZE;
2745 0 : CHK_RET(ConstructPayLoadWqe(
2746 : const_cast<void*>(remoteAddr), const_cast<void*>(localAddr), length, wqeType, aux, wqeInfoVec,
2747 : txSendDataTimes));
2748 : }
2749 :
2750 0 : u32 maxLength = 0;
2751 0 : for (u32 i = 0; i < wqeInfoVec.size(); i++) {
2752 0 : if (wqeInfoVec[i].wqeData.memList.len > maxLength) {
2753 0 : maxLength = wqeInfoVec[i].wqeData.memList.len;
2754 : }
2755 : }
2756 :
2757 0 : u32 actualMultiQpNum = GetActualQpNum(maxLength);
2758 :
2759 0 : HCCL_DEBUG(
2760 : "[TransportIbverbs][TxSendDataAndNotify] UseMultiQp[%d] MultiQpNum[%u] actualMultiQpNum[%u] maxLength[%u]",
2761 : UseMultiQp(), qpsPerConnection_, actualMultiQpNum, maxLength);
2762 0 : if (UseMultiQp() && actualMultiQpNum != 1 && actualMultiQpNum <= qpsPerConnection_ && maxLength != 0) {
2763 0 : std::vector<std::vector<WqeInfo>> multiQpWqeInfoVct(actualMultiQpNum, wqeInfoVec);
2764 0 : for (u32 i = 0; i < wqeInfoVec.size(); i++) {
2765 0 : WqeInfo tmpWqeInfo = wqeInfoVec[i];
2766 0 : u32 curLen = tmpWqeInfo.wqeData.memList.len;
2767 0 : std::vector<u32> splittedLen = RdmaLengthSplit(curLen, actualMultiQpNum);
2768 0 : uint64_t curSrcAddr = tmpWqeInfo.wqeData.memList.addr;
2769 0 : uint64_t curDstAddr = tmpWqeInfo.wqeData.dstAddr;
2770 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
2771 0 : multiQpWqeInfoVct[qpIndex][i].wqeData.memList.len = splittedLen[qpIndex];
2772 0 : multiQpWqeInfoVct[qpIndex][i].wqeData.memList.addr = curSrcAddr;
2773 0 : multiQpWqeInfoVct[qpIndex][i].wqeData.dstAddr = curDstAddr;
2774 0 : curSrcAddr += splittedLen[qpIndex];
2775 0 : curDstAddr += splittedLen[qpIndex];
2776 : }
2777 0 : }
2778 :
2779 : // useOneDoorbell 配置成true。最后一个payload去按doorbell
2780 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
2781 0 : CHK_RET(RdmaSendAsync(
2782 : multiQpWqeInfoVct[qpIndex], stream, true, qpIndex)); // 多QP使用同一个stream异步doorbell触发
2783 : }
2784 0 : } else {
2785 0 : if (machinePara_.nicDeploy == NICDeployment::NIC_DEPLOYMENT_DEVICE) {
2786 0 : CHK_RET(RdmaSendAsync(wqeInfoVec, stream, GetUseOneDoorbellValue()));
2787 : } else {
2788 0 : CHK_RET(RdmaSendAsyncHostNIC(wqeInfoVec, stream));
2789 : }
2790 : }
2791 0 : return HCCL_SUCCESS;
2792 0 : }
2793 :
2794 : HcclResult
2795 0 : TransportIbverbs::WriteAsync(struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, Stream& stream)
2796 : {
2797 0 : struct WrAuxInfo aux = {};
2798 0 : return WriteCommon(remoteBuf.addr, localBuf.addr, remoteBuf.size, stream, WqeType::WQE_TYPE_DATA, aux);
2799 : }
2800 :
2801 0 : HcclResult TransportIbverbs::WriteReduceAsync(
2802 : struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, const HcclDataType datatype,
2803 : HcclReduceOp redOp, Stream& stream)
2804 : {
2805 0 : struct WrAuxInfo aux = {};
2806 0 : aux.dataType = RDMA_REDUCE_DATA_TYPE_TABLE[datatype];
2807 0 : aux.reduceType = RDMA_REDUCE_OP_TYPE_TABLE[redOp];
2808 0 : if (aux.dataType == static_cast<uint8_t>(RdmaReduceDataType::RDMA_REDUCE_DATA_INVALID)
2809 0 : || aux.reduceType == static_cast<uint8_t>(RdmaReduceOpType::RDMA_REDUCE_OP_INVALID)) {
2810 0 : HCCL_ERROR(
2811 : "unsupported data type [%s] or Reduce type [%s]", GetDataTypeEnumStr(datatype).c_str(),
2812 : GetReduceOpEnumStr(redOp).c_str());
2813 0 : return HCCL_E_INTERNAL;
2814 : }
2815 :
2816 0 : return WriteCommon(remoteBuf.addr, localBuf.addr, remoteBuf.size, stream, WqeType::WQE_TYPE_DATA_WITH_REDUCE, aux);
2817 : }
2818 :
2819 0 : HcclResult TransportIbverbs::WriteSync(
2820 : [[maybe_unused]] struct Transport::Buffer& remoteBuf, [[maybe_unused]] struct Transport::Buffer& localBuf,
2821 : [[maybe_unused]] Stream& stream)
2822 : {
2823 0 : return HCCL_E_NOT_SUPPORT;
2824 : }
2825 :
2826 : HcclResult
2827 0 : TransportIbverbs::ReadAsync(struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, Stream& stream)
2828 : {
2829 0 : struct WrAuxInfo aux = {};
2830 0 : return WriteCommon(remoteBuf.addr, localBuf.addr, remoteBuf.size, stream, WqeType::WQE_TYPE_READ_DATA, aux);
2831 : }
2832 :
2833 0 : HcclResult TransportIbverbs::ReadSync(
2834 : [[maybe_unused]] struct Transport::Buffer& localBuf, [[maybe_unused]] struct Transport::Buffer& remoteBuf,
2835 : [[maybe_unused]] Stream& stream)
2836 : {
2837 0 : return HCCL_E_NOT_SUPPORT;
2838 : }
2839 :
2840 0 : HcclResult TransportIbverbs::GetLocalNotify(std::vector<HcclSignalInfo>& localNotify)
2841 : {
2842 : HcclSignalInfo notifyInfo;
2843 0 : CHK_SMART_PTR_NULL(dataNotify_);
2844 0 : CHK_RET(dataNotify_->GetNotifyData(notifyInfo));
2845 0 : localNotify.push_back(notifyInfo);
2846 :
2847 0 : CHK_SMART_PTR_NULL(ackNotify_);
2848 0 : CHK_RET(ackNotify_->GetNotifyData(notifyInfo));
2849 0 : localNotify.push_back(notifyInfo);
2850 :
2851 0 : CHK_SMART_PTR_NULL(dataAckNotify_);
2852 0 : CHK_RET(dataAckNotify_->GetNotifyData(notifyInfo));
2853 0 : localNotify.push_back(notifyInfo);
2854 :
2855 : // 提取新增的notify资源
2856 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
2857 0 : for (u32 j = 0; j < notifyNum_; j++) {
2858 0 : CHK_SMART_PTR_NULL(userMultiQpLocalNotify_[i][j]);
2859 0 : CHK_RET(userMultiQpLocalNotify_[i][j]->GetNotifyData(notifyInfo));
2860 0 : localNotify.push_back(notifyInfo);
2861 : }
2862 : }
2863 :
2864 0 : return HCCL_SUCCESS;
2865 : }
2866 :
2867 14 : HcclResult TransportIbverbs::GetTransportErrorCqe(
2868 : const HcclNetDevCtx netDevCtx, std::vector<std::pair<TransportBase*, CqeInfo>>& infos, u32& num)
2869 : {
2870 14 : if (g_qpn2IbversLinkMap_.Size() == 0) {
2871 14 : num = 0;
2872 14 : return HCCL_SUCCESS;
2873 : }
2874 :
2875 0 : if (UNLIKELY(!g_flag)) {
2876 0 : CHK_RET(IsSuppCqeErrInfoListConfig(g_isSupCqeErrInfoListConfig));
2877 0 : g_flag = true;
2878 : }
2879 :
2880 0 : CHK_PTR_NULL(netDevCtx);
2881 0 : s32 deviceLogicId = (static_cast<NetDevContext*>(netDevCtx))->GetLogicId();
2882 0 : s32 devicePhyId = (static_cast<NetDevContext*>(netDevCtx))->GetPhyId();
2883 0 : HcclIpAddress localIp = (static_cast<NetDevContext*>(netDevCtx))->GetLocalIp();
2884 0 : NicType nicType = (static_cast<NetDevContext*>(netDevCtx))->GetNicType();
2885 0 : CHK_PRT_RET(
2886 : nicType == NicType::HOST_NIC_TYPE,
2887 : HCCL_WARNING("[TransportIbverbs][GetTransportErrorCqe] nicType[%d] not support", nicType), HCCL_SUCCESS);
2888 0 : RaResourceInfo raResourceInfo;
2889 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId).GetRaResourceInfo(raResourceInfo));
2890 0 : RdmaHandle rdmaHandle = raResourceInfo.nicSocketMap[localIp].nicRdmaHandle;
2891 0 : CHK_PTR_NULL(rdmaHandle);
2892 :
2893 0 : if (g_isSupCqeErrInfoListConfig) {
2894 0 : u32 loop = 0;
2895 0 : if (num > CQE_ARRAY_SIZE) {
2896 0 : loop = (num % CQE_ARRAY_SIZE) ? (num / CQE_ARRAY_SIZE) : ((num / CQE_ARRAY_SIZE) - 1);
2897 : }
2898 :
2899 0 : struct CqeErrInfo infolist[CQE_ARRAY_SIZE] = {};
2900 0 : u32 cqeNum = CQE_ARRAY_SIZE;
2901 0 : for (u32 index = 0; index <= loop; index++) {
2902 0 : cqeNum = (index == loop) ? (num - index * CQE_ARRAY_SIZE) : CQE_ARRAY_SIZE;
2903 0 : u32 temNum = cqeNum;
2904 0 : CHK_RET(hrtRaGetCqeErrInfoList(rdmaHandle, infolist, &temNum));
2905 0 : ProcessCqeInfo(devicePhyId, infolist, temNum, infos);
2906 0 : if (temNum < cqeNum) {
2907 0 : break;
2908 : }
2909 : }
2910 : } else {
2911 0 : struct CqeErrInfo infolist[1] = {};
2912 0 : CHK_RET(hrtRaGetCqeErrInfo(devicePhyId, &infolist[0]));
2913 0 : if (infolist[0].status == 0) {
2914 0 : num = 0;
2915 0 : return HCCL_SUCCESS;
2916 : }
2917 0 : u32 cqeNum = 1;
2918 0 : ProcessCqeInfo(devicePhyId, infolist, cqeNum, infos);
2919 : }
2920 :
2921 0 : num = infos.size();
2922 :
2923 0 : return HCCL_SUCCESS;
2924 0 : }
2925 :
2926 0 : void TransportIbverbs::ProcessCqeInfo(
2927 : const s32 deviceId, const struct CqeErrInfo* infolist, const u32 cqeNum,
2928 : std::vector<std::pair<TransportBase*, CqeInfo>>& infos)
2929 : {
2930 0 : for (u32 i = 0; i < cqeNum; i++) {
2931 : // localPhyId + qpn
2932 0 : auto it = g_qpn2IbversLinkMap_.Find(((static_cast<u64>(deviceId) << DEV_PHY_ID_BIT) | infolist[i].qpn));
2933 0 : if (it.second) {
2934 0 : TransportBase* ptr = reinterpret_cast<TransportBase*>(it.first->second);
2935 0 : infos.push_back(
2936 0 : std::make_pair(ptr, CqeInfo(infolist[i].time, infolist[i].status, it.first->second->GetRemoteIp())));
2937 0 : cqeErrQpn_ = infolist[i].qpn;
2938 : } else {
2939 0 : HCCL_RUN_WARNING("[GetTransportErrorCqe]get err failed, transport is not find.");
2940 : }
2941 : }
2942 0 : return;
2943 : }
2944 :
2945 0 : HcclIpAddress& TransportIbverbs::GetRemoteIp() { return machinePara_.remoteIpAddr; }
2946 :
2947 0 : HcclResult TransportIbverbs::GetTransportId(u32& id)
2948 : {
2949 0 : id = cqeErrQpn_;
2950 0 : return HCCL_SUCCESS;
2951 : }
2952 :
2953 0 : HcclResult TransportIbverbs::ExchangeCapabilityHybrid()
2954 : {
2955 0 : HCCL_INFO("[Hybrid][TransportIbverbs] Starting capability exchange");
2956 :
2957 : // 1. 构造本地能力信息（使用公共头文件中的默认值）
2958 : using namespace hcomm;
2959 : RoCECapability localCap;
2960 0 : localCap.InitDefaults();
2961 0 : localCap.nicDeploy = NICDeployment::NIC_DEPLOYMENT_DEVICE;
2962 0 : localCap.commStack = CommStackType::COMM_STACK_TRANSPORT_IBVERBS;
2963 :
2964 : // 2. 发送本地能力（合并为单次发送：totalLength已包含结构体大小）
2965 0 : CHK_RET(defaultSocket_->Send(&localCap, sizeof(localCap)));
2966 0 : HCCL_INFO("[Hybrid][TransportIbverbs] Sent capability, version=%u", localCap.version);
2967 :
2968 : // 3. 接收对端能力（单次接收）
2969 : RoCECapability recvCap;
2970 0 : CHK_RET(defaultSocket_->Recv(&recvCap, sizeof(recvCap)));
2971 :
2972 : // 4. 先检查魔数，如果不对可能是旧版本，需要回退
2973 0 : if (!RoCECapability::CheckMagic(reinterpret_cast<uint8_t*>(&recvCap), sizeof(recvCap))) {
2974 0 : HCCL_WARNING("[Hybrid][TransportIbverbs] Magic mismatch, peer may be old version. "
2975 : "Falling back to native mode.");
2976 : // 回退到原生模式
2977 0 : isHybridMode_ = false;
2978 0 : return HCCL_SUCCESS;
2979 : }
2980 :
2981 : // 5. 魔数正确，解析对端能力
2982 : RoCECapability remoteCap;
2983 0 : if (!remoteCap.Deserialize(reinterpret_cast<uint8_t*>(&recvCap), sizeof(recvCap))) {
2984 0 : HCCL_ERROR("[Hybrid][TransportIbverbs] Failed to deserialize capability");
2985 0 : return HCCL_E_PARA;
2986 : }
2987 :
2988 : // 6. 校验字段有效性
2989 0 : if (!remoteCap.Validate()) {
2990 0 : HCCL_ERROR("[Hybrid][TransportIbverbs] Capability validation failed");
2991 0 : return HCCL_E_INTERNAL;
2992 : }
2993 :
2994 : // 7. 版本兼容性处理（高版本兼容低版本）
2995 0 : if (remoteCap.version > ROCE_CAPABILITY_VERSION) {
2996 : // 对端版本更高，使用本地版本的功能集（最小公分母）
2997 0 : HCCL_INFO(
2998 : "[Hybrid][TransportIbverbs] Remote version %u > local %u, using local version features", remoteCap.version,
2999 : ROCE_CAPABILITY_VERSION);
3000 0 : } else if (remoteCap.version < ROCE_CAPABILITY_VERSION) {
3001 : // 对端版本更低，使用对端版本的功能集（向下兼容）
3002 0 : HCCL_INFO(
3003 : "[Hybrid][TransportIbverbs] Remote version %u < local %u, using remote version features", remoteCap.version,
3004 : ROCE_CAPABILITY_VERSION);
3005 : }
3006 :
3007 0 : isHybridMode_ = (remoteCap.commStack == CommStackType::COMM_STACK_HOST_CPU_ROCE) ? true : false;
3008 0 : HCCL_INFO(
3009 : "[Hybrid][TransportIbverbs]Exchange mode success, Remote is %s",
3010 : isHybridMode_ ? "HostCpuRoceChannel" : "TransportIbverbs");
3011 :
3012 0 : return HCCL_SUCCESS;
3013 : }
3014 :
3015 0 : HcclResult TransportIbverbs::GetDrainRemSrcMem(void*& remoteAddr, uint32_t& remoteKey, uint32_t& size)
3016 : {
3017 0 : auto opType = static_cast<u32>(MemType::NOTIFY_SRC_MEM);
3018 0 : remoteAddr = remoteMemMsg_[opType].addr;
3019 0 : remoteKey = remoteMemMsg_[opType].lkey;
3020 0 : size = remoteMemMsg_[opType].len;
3021 0 : return HCCL_SUCCESS;
3022 : }
3023 :
3024 1 : HcclResult TransportIbverbs::Drain(Stream& stream)
3025 : {
3026 1 : auto remoteType = static_cast<u32>(MemType::NOTIFY_SRC_MEM);
3027 1 : auto localType = static_cast<u32>(MemType::DATA_NOTIFY_MEM);
3028 1 : CHK_PTR_NULL(memMsg_[localType].addr);
3029 1 : CHK_PTR_NULL(remoteMemMsg_[remoteType].addr);
3030 1 : CHK_RET(Fence());
3031 1 : std::vector<WqeInfo> wqeInfoVec;
3032 1 : WrAuxInfo aux = {};
3033 1 : CHK_RET(AddWqeList(
3034 : remoteMemMsg_[remoteType].addr, memMsg_[localType].addr, memMsg_[localType].len, WqeType::WQE_TYPE_READ_DATA,
3035 : aux, wqeInfoVec));
3036 1 : CHK_RET(RdmaSendAsync(wqeInfoVec, stream, false));
3037 0 : CHK_SMART_PTR_NULL(dataNotify_);
3038 0 : CHK_RET(LocalIpcNotify::Wait(
3039 : stream, dispatcher_, dataNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME, machinePara_.localUserrank,
3040 : machinePara_.remoteWorldRank));
3041 0 : return HCCL_SUCCESS;
3042 1 : }
3043 :
3044 0 : HcclResult TransportIbverbs::GetDrainLocalDataNotify(void*& localAddr, uint32_t& lkey, HcclSignalInfo& dataNotify)
3045 : {
3046 0 : localAddr = memMsg_[DATA_NOTIFY_MEM].addr;
3047 0 : lkey = memMsg_[DATA_NOTIFY_MEM].lkey;
3048 0 : CHK_SMART_PTR_NULL(dataNotify_);
3049 0 : CHK_RET(dataNotify_->GetNotifyData(dataNotify));
3050 0 : return HCCL_SUCCESS;
3051 : }
3052 : } // namespace hccl
|
