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