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