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 <chrono>
14 : #include <memory>
15 : #include "network/hccp_common.h"
16 : #include "device_capacity.h"
17 : #include "dlhns_function.h"
18 : #include "adapter_verbs.h"
19 : #include "transport_device_ibverbs.h"
20 : #include "new/hccl_dispatcher_ctx.h"
21 :
22 : constexpr u32 RDMA_QP_EXPECT_STATUS_PAUSE = 5;
23 : constexpr u32 RDMA_QP_EXPECT_STATUS_CONNECTED = 1;
24 : constexpr s32 RDMA_QP_NO_MEM = -12;
25 :
26 : constexpr u32 RDMA_WRITE_NOTIFY_OFFSET_MASK = 0xffffff;
27 : constexpr u32 RDMA_WRITE_NOTIFY_VALUE_RECORD = 0x1000000;
28 :
29 : // 内存屏障，确保wqe下到HBM里
30 : #if defined(__x86_64__)
31 : #define HCOMM_DSB() asm volatile("" ::: "memory")
32 : #elif defined(__aarch64__)
33 : #define HCOMM_DSB() asm volatile("dsb st" ::: "memory")
34 : #else
35 : #define HCOMM_DSB()
36 : #endif
37 :
38 : namespace hccl {
39 : namespace {
40 11 : inline BufferKey<uintptr_t, u64> MakeMemLookupKey(u64 logicalStartVa, u64 size)
41 : {
42 11 : return BufferKey<uintptr_t, u64>(static_cast<uintptr_t>(logicalStartVa), size);
43 : }
44 :
45 3 : inline BufferKey<uintptr_t, u64> MakeMemLookupKey(const void* logicalPtr, u64 size)
46 : {
47 3 : return MakeMemLookupKey(static_cast<u64>(reinterpret_cast<uintptr_t>(logicalPtr)), size);
48 : }
49 :
50 2 : inline void* LogicalPtrToDevPtr(const RoceMemDetails& md, const void* logicalPtr)
51 : {
52 2 : const u64 logicalVa = static_cast<u64>(reinterpret_cast<uintptr_t>(logicalPtr));
53 2 : const u64 offset = logicalVa - md.addr;
54 2 : const u64 devVa = md.devAddr + offset;
55 2 : return reinterpret_cast<void*>(static_cast<uintptr_t>(devVa));
56 : }
57 : } // namespace
58 :
59 : std::atomic<u64> TransportDeviceIbverbs::wrIdOffset_ = {0};
60 :
61 10 : TransportDeviceIbverbs::TransportDeviceIbverbs(
62 : DispatcherPub* dispatcher, const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara,
63 10 : std::chrono::milliseconds timeout, const TransportDeviceIbverbsData& transDevIbverbsData)
64 : : TransportIbverbs(dispatcher, notifyPool, machinePara, timeout),
65 10 : transDevIbverbsData_(transDevIbverbsData)
66 10 : {}
67 :
68 11 : TransportDeviceIbverbs::~TransportDeviceIbverbs()
69 : {
70 10 : HCCL_DEBUG("~TransportDeviceIbverbs Enter!");
71 :
72 10 : (void)DeInit();
73 :
74 10 : if (machinePara_.deviceLogicId >= 0 && (static_cast<u32>(machinePara_.deviceLogicId) < MAX_MODULE_DEVICE_NUM)) {
75 10 : if (instanceRef_[machinePara_.deviceLogicId].Unref() == 0) {
76 3 : std::unique_lock<std::mutex> lock(notifyValueMutex_[machinePara_.deviceLogicId]);
77 3 : notifyValueMem_[machinePara_.deviceLogicId].free();
78 3 : }
79 : }
80 10 : HCCL_DEBUG("~TransportDeviceIbverbs Success!");
81 11 : }
82 :
83 1 : HcclResult TransportDeviceIbverbs::InitDrainNotifyInfo()
84 : {
85 1 : HCCL_DEBUG(
86 : "[%s] RemoteNotifyAddr[%llu], remoteNotifyKey[%u], localDataNotifyAddr[%llu], localDataNotifyKey[%u],"
87 : "notifySize[%u]",
88 : __func__, transDevIbverbsData_.remoteNotifyValueAddr, transDevIbverbsData_.remoteNotifyValueKey,
89 : transDevIbverbsData_.localDataNotifyAddr, transDevIbverbsData_.localDataNotifyKey,
90 : transDevIbverbsData_.notifySize);
91 1 : CHK_PRT_RET(
92 : (transDevIbverbsData_.localDataNotifyAddr == 0 || transDevIbverbsData_.remoteNotifyValueAddr == 0),
93 : HCCL_ERROR(
94 : "[%s] Notify addr is nullptr, RemoteNotifyAddr[%llu], localDataNotifyAddr[%llu]",
95 : transDevIbverbsData_.remoteNotifyValueAddr, transDevIbverbsData_.localDataNotifyAddr),
96 : HCCL_E_PTR);
97 1 : memMsg_[MemType::DATA_NOTIFY_MEM].addr = reinterpret_cast<void*>(transDevIbverbsData_.localDataNotifyAddr);
98 1 : memMsg_[MemType::DATA_NOTIFY_MEM].lkey = transDevIbverbsData_.localDataNotifyKey;
99 1 : memMsg_[MemType::DATA_NOTIFY_MEM].len = transDevIbverbsData_.notifySize;
100 1 : remoteMemMsg_[MemType::NOTIFY_SRC_MEM].addr = reinterpret_cast<void*>(transDevIbverbsData_.remoteNotifyValueAddr);
101 1 : remoteMemMsg_[MemType::NOTIFY_SRC_MEM].lkey = transDevIbverbsData_.remoteNotifyValueKey;
102 1 : remoteMemMsg_[MemType::NOTIFY_SRC_MEM].len = transDevIbverbsData_.notifySize;
103 1 : CHK_RET(SignalInit(transDevIbverbsData_.dataNotify, dataNotify_));
104 1 : return HCCL_SUCCESS;
105 : }
106 :
107 6 : HcclResult TransportDeviceIbverbs::Init()
108 : {
109 6 : HCCL_DEBUG("TransportDeviceIbverbs Init Enter! notifyNum[%u]", machinePara_.notifyNum);
110 6 : if (transDevIbverbsData_.useMemDetailsMgr) {
111 6 : return InitMemDetails();
112 : }
113 0 : CHK_RET(SignalInit(transDevIbverbsData_.ackNotify, ackNotify_));
114 0 : CHK_RET(SignalInit(transDevIbverbsData_.dataNotify, dataNotify_));
115 0 : CHK_RET(SignalInit(transDevIbverbsData_.dataAckNotify, dataAckNotify_));
116 0 : constexpr u32 QPINFO_SIZE_MAX = 33;
117 0 : constexpr u32 QPINFO_SIZE_MIN = 1;
118 0 : constexpr u32 QP_PERCONNECTION_MAX = 32;
119 0 : constexpr u32 QP_PERCONNECTION_MIN = 1;
120 0 : u32 qpInfoSize = transDevIbverbsData_.qpInfo.size();
121 0 : if (transDevIbverbsData_.qpsPerConnection + static_cast<u32>(qpInfoSize > 1) != qpInfoSize
122 0 : || qpInfoSize > QPINFO_SIZE_MAX || qpInfoSize < QPINFO_SIZE_MIN
123 0 : || transDevIbverbsData_.qpsPerConnection > QP_PERCONNECTION_MAX
124 0 : || transDevIbverbsData_.qpsPerConnection < QP_PERCONNECTION_MIN) {
125 0 : HCCL_ERROR(
126 : "[TransportDeviceIbverbs][Init]QPNum[%d] or qpInfos size[%u] is invalid",
127 : transDevIbverbsData_.qpsPerConnection, qpInfoSize);
128 0 : return HCCL_E_INTERNAL;
129 : }
130 0 : combineAiQpInfo_.aiQpInfo.aiQpAddr = transDevIbverbsData_.qpInfo[0].qpPtr;
131 0 : combineAiQpInfo_.aiQpInfo.sqIndex = transDevIbverbsData_.qpInfo[0].sqIndex;
132 0 : combineAiQpInfo_.aiQpInfo.dbIndex = transDevIbverbsData_.qpInfo[0].dbIndex;
133 0 : combineAiQpInfos_.resize(transDevIbverbsData_.qpsPerConnection);
134 0 : for (u32 i = 1, j = 0; i < qpInfoSize; i++, j++) {
135 0 : combineAiQpInfos_[j].aiQpInfo.aiQpAddr = transDevIbverbsData_.qpInfo[i].qpPtr;
136 0 : combineAiQpInfos_[j].aiQpInfo.sqIndex = transDevIbverbsData_.qpInfo[i].sqIndex;
137 0 : combineAiQpInfos_[j].aiQpInfo.dbIndex = transDevIbverbsData_.qpInfo[i].dbIndex;
138 0 : HCCL_DEBUG(
139 : "TransportDeviceIbverbs Init multiQp[%u], aiQpAddr[%llu] sqIndex[%u] dbIndex[%u]", j,
140 : transDevIbverbsData_.qpInfo[i].qpPtr, transDevIbverbsData_.qpInfo[i].sqIndex,
141 : transDevIbverbsData_.qpInfo[i].dbIndex);
142 : }
143 0 : notifySize_ = transDevIbverbsData_.notifySize;
144 0 : remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].addr = transDevIbverbsData_.inputBufferPtr;
145 0 : remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].lkey = transDevIbverbsData_.remoteInputKey;
146 :
147 0 : remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].addr = transDevIbverbsData_.outputBufferPtr;
148 0 : remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].lkey = transDevIbverbsData_.remoteOutputKey;
149 :
150 0 : u32 ackNotifyIdx = static_cast<u32>(MemType::ACK_NOTIFY_MEM);
151 0 : remoteMemMsg_[ackNotifyIdx].addr = reinterpret_cast<void*>(transDevIbverbsData_.remoteAckNotifyDetails.addr);
152 0 : remoteMemMsg_[ackNotifyIdx].notifyId = transDevIbverbsData_.remoteAckNotifyDetails.notifyId;
153 0 : remoteMemMsg_[ackNotifyIdx].lkey = transDevIbverbsData_.remoteAckNotifyDetails.key;
154 :
155 0 : u32 dataNotifyIdx = static_cast<u32>(MemType::DATA_NOTIFY_MEM);
156 0 : remoteMemMsg_[dataNotifyIdx].addr = reinterpret_cast<void*>(transDevIbverbsData_.remoteDataNotifyDetails.addr);
157 0 : remoteMemMsg_[dataNotifyIdx].notifyId = transDevIbverbsData_.remoteDataNotifyDetails.notifyId;
158 0 : remoteMemMsg_[dataNotifyIdx].lkey = transDevIbverbsData_.remoteDataNotifyDetails.key;
159 :
160 0 : u32 dataAckIdx = static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM);
161 0 : remoteMemMsg_[dataAckIdx].addr = reinterpret_cast<void*>(transDevIbverbsData_.remoteDataAckNotifyDetails.addr);
162 0 : remoteMemMsg_[dataAckIdx].notifyId = transDevIbverbsData_.remoteDataAckNotifyDetails.notifyId;
163 0 : remoteMemMsg_[dataAckIdx].lkey = transDevIbverbsData_.remoteDataAckNotifyDetails.key;
164 :
165 0 : HCCL_INFO(
166 : "%s ACK:addr[0x%llx] notifyId[%d] lkey[%u], DATA:addr[0x%llx] notifyId[%d] lkey[%u], "
167 : "DATA_ACK:addr[0x%llx] notifyId[%d] lkey[%u]",
168 : __func__, remoteMemMsg_[ackNotifyIdx].addr, remoteMemMsg_[ackNotifyIdx].notifyId,
169 : remoteMemMsg_[ackNotifyIdx].lkey, remoteMemMsg_[dataNotifyIdx].addr, remoteMemMsg_[dataNotifyIdx].notifyId,
170 : remoteMemMsg_[dataNotifyIdx].lkey, remoteMemMsg_[dataAckIdx].addr, remoteMemMsg_[dataAckIdx].notifyId,
171 : remoteMemMsg_[dataAckIdx].lkey);
172 :
173 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr
174 0 : = reinterpret_cast<void*>(transDevIbverbsData_.localNotifyValueAddr);
175 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey = transDevIbverbsData_.notifyValueKey;
176 0 : localInputMem_ = transDevIbverbsData_.localInputMem;
177 0 : memMsg_[MemType::USER_INPUT_MEM].addr = reinterpret_cast<void*>(transDevIbverbsData_.localInputMem.addr);
178 0 : memMsg_[MemType::USER_INPUT_MEM].len = transDevIbverbsData_.localInputMem.size;
179 0 : memMsg_[MemType::USER_INPUT_MEM].lkey = transDevIbverbsData_.localInputMem.key;
180 :
181 0 : localOutputMem_ = transDevIbverbsData_.localOutputMem;
182 0 : memMsg_[MemType::USER_OUTPUT_MEM].addr = reinterpret_cast<void*>(transDevIbverbsData_.localOutputMem.addr);
183 0 : memMsg_[MemType::USER_OUTPUT_MEM].len = transDevIbverbsData_.localOutputMem.size;
184 0 : memMsg_[MemType::USER_OUTPUT_MEM].lkey = transDevIbverbsData_.localOutputMem.key;
185 :
186 0 : notifyValueAddr_ = reinterpret_cast<void*>(transDevIbverbsData_.localNotifyValueAddr);
187 0 : CHK_RET(CheckDeviceId());
188 0 : CHK_RET(DlHnsFunction::GetInstance().DlHnsFunctionInit());
189 0 : transportAttr_.linkType = LinkType::LINK_ROCE;
190 0 : multiQpThreshold_ = transDevIbverbsData_.multiQpThreshold;
191 0 : qpsPerConnection_ = transDevIbverbsData_.qpsPerConnection;
192 0 : if (transDevIbverbsData_.userLocalNotify.size() != qpsPerConnection_
193 0 : || transDevIbverbsData_.userRemoteNotifyDetails.size() != qpsPerConnection_) {
194 0 : HCCL_ERROR(
195 : "[TransportDeviceIbverbs][Init]userLocalNotify size[%u] is not equal to qpsPerConnection[%u]",
196 : transDevIbverbsData_.userLocalNotify.size(), qpsPerConnection_);
197 0 : return HCCL_E_INTERNAL;
198 : }
199 :
200 0 : userMultiQpLocalNotify_.resize(transDevIbverbsData_.qpsPerConnection);
201 0 : u32 multiQpExtNotifyLength = transDevIbverbsData_.qpsPerConnection > 1 ? transDevIbverbsData_.qpsPerConnection : 0;
202 0 : multiQpDataNotify_.resize(multiQpExtNotifyLength);
203 0 : for (u32 i = 0; i < transDevIbverbsData_.qpsPerConnection; ++i) {
204 0 : CHK_PRT_RET(
205 : transDevIbverbsData_.userLocalNotify[i].empty() && transDevIbverbsData_.qpsPerConnection > 1,
206 : HCCL_ERROR(
207 : "[TransportDeviceIbverbs][Init]userLocalNotify[%u] is empty, qpsPerConnection[%u]", i,
208 : transDevIbverbsData_.qpsPerConnection),
209 : HCCL_E_INTERNAL);
210 0 : u32 singleQpNotifyNum = transDevIbverbsData_.qpsPerConnection > 1 ?
211 0 : transDevIbverbsData_.userLocalNotify[i].size() - 1 :
212 0 : transDevIbverbsData_.userLocalNotify[i].size();
213 0 : CHK_PRT_RET(
214 : singleQpNotifyNum != notifyNum_,
215 : HCCL_ERROR(
216 : "[TransportDeviceIbverbs][Init] qpIdx[%u] userLocalNotify notifynum[%u] is not equal to notifyNum_[%u]",
217 : i, singleQpNotifyNum, notifyNum_),
218 : HCCL_E_INTERNAL);
219 0 : userMultiQpLocalNotify_[i].resize(singleQpNotifyNum);
220 0 : for (u32 j = 0; j < singleQpNotifyNum; ++j) {
221 0 : CHK_RET(SignalInit(transDevIbverbsData_.userLocalNotify[i][j], userMultiQpLocalNotify_[i][j]));
222 : }
223 0 : if (transDevIbverbsData_.qpsPerConnection > 1) {
224 0 : CHK_RET(SignalInit(transDevIbverbsData_.userLocalNotify[i][singleQpNotifyNum], multiQpDataNotify_[i]));
225 : }
226 : }
227 :
228 0 : userMultiQpRemoteNotifyMsg_.resize(transDevIbverbsData_.qpsPerConnection);
229 0 : multiQpDataNotifyRemoteMemMsg_.resize(multiQpExtNotifyLength);
230 0 : for (u32 i = 0; i < transDevIbverbsData_.qpsPerConnection; ++i) {
231 0 : CHK_PRT_RET(
232 : transDevIbverbsData_.userRemoteNotifyDetails[i].empty() && transDevIbverbsData_.qpsPerConnection > 1,
233 : HCCL_ERROR(
234 : "[TransportDeviceIbverbs][Init]userLocalNotify[%u] is empty, qpsPerConnection[%u]", i,
235 : transDevIbverbsData_.qpsPerConnection),
236 : HCCL_E_INTERNAL);
237 0 : u32 singleQpNotifyNum = transDevIbverbsData_.qpsPerConnection > 1 ?
238 0 : transDevIbverbsData_.userRemoteNotifyDetails[i].size() - 1 :
239 0 : transDevIbverbsData_.userRemoteNotifyDetails[i].size();
240 0 : CHK_PRT_RET(
241 : singleQpNotifyNum != notifyNum_,
242 : HCCL_ERROR(
243 : "[TransportDeviceIbverbs][Init] qpIdx[%u] userLocalNotify notifynum[%u] is not equal to notifyNum_[%u]",
244 : i, singleQpNotifyNum, notifyNum_),
245 : HCCL_E_INTERNAL);
246 0 : userMultiQpRemoteNotifyMsg_[i].resize(singleQpNotifyNum);
247 0 : u32 j = 0;
248 0 : for (; j < singleQpNotifyNum; ++j) {
249 0 : userMultiQpRemoteNotifyMsg_[i][j].addr
250 0 : = reinterpret_cast<void*>(transDevIbverbsData_.userRemoteNotifyDetails[i][j].addr);
251 0 : userMultiQpRemoteNotifyMsg_[i][j].notifyId = transDevIbverbsData_.userRemoteNotifyDetails[i][j].notifyId;
252 0 : userMultiQpRemoteNotifyMsg_[i][j].lkey = transDevIbverbsData_.userRemoteNotifyDetails[i][j].key;
253 0 : HCCL_INFO(
254 : "userMultiQpRemoteNotifyMsg_[%u][%u] addr[0x%llx] notifyId[%u] lkey[%u]", i, j,
255 : userMultiQpRemoteNotifyMsg_[i][j].addr, userMultiQpRemoteNotifyMsg_[i][j].notifyId,
256 : userMultiQpRemoteNotifyMsg_[i][j].lkey);
257 : }
258 0 : if (transDevIbverbsData_.qpsPerConnection > 1) {
259 0 : multiQpDataNotifyRemoteMemMsg_[i].addr
260 0 : = reinterpret_cast<void*>(transDevIbverbsData_.userRemoteNotifyDetails[i][j].addr);
261 0 : multiQpDataNotifyRemoteMemMsg_[i].notifyId = transDevIbverbsData_.userRemoteNotifyDetails[i][j].notifyId;
262 0 : multiQpDataNotifyRemoteMemMsg_[i].lkey = transDevIbverbsData_.userRemoteNotifyDetails[i][j].key;
263 0 : HCCL_INFO(
264 : "multiQpDataNotifyRemoteMemMsg_[%u] addr[0x%llx] notifyId[%u] lkey[%u]", i,
265 : multiQpDataNotifyRemoteMemMsg_[i].addr, multiQpDataNotifyRemoteMemMsg_[i].notifyId,
266 : multiQpDataNotifyRemoteMemMsg_[i].lkey);
267 : }
268 : }
269 0 : useAtomicWrite_ = transDevIbverbsData_.useAtomicWrite;
270 0 : HCCL_USER_CRITICAL_LOG(
271 : "create hccl transport:communicator[%s], local rank[%u], remote rank[%u],"
272 : "transporttype[%s], atomicWrite[%d]",
273 : machinePara_.tag.c_str(), machinePara_.localUserrank, machinePara_.remoteUserrank,
274 : GetLinkTypeEnumStr(GetLinkType()).c_str(), useAtomicWrite_);
275 :
276 0 : return HCCL_SUCCESS;
277 : }
278 :
279 6 : HcclResult TransportDeviceIbverbs::InitMemDetails()
280 : {
281 6 : constexpr u32 QPINFO_SIZE_MAX = 33;
282 6 : constexpr u32 QPINFO_SIZE_MIN = 1;
283 6 : constexpr u32 QP_PERCONNECTION_MAX = 32;
284 6 : constexpr u32 QP_PERCONNECTION_MIN = 1;
285 6 : u32 qpSize = transDevIbverbsData_.qpInfo.size();
286 6 : if (transDevIbverbsData_.qpsPerConnection + static_cast<u32>(qpSize > 1) != qpSize || qpSize > QPINFO_SIZE_MAX
287 5 : || qpSize < QPINFO_SIZE_MIN || transDevIbverbsData_.qpsPerConnection > QP_PERCONNECTION_MAX
288 5 : || transDevIbverbsData_.qpsPerConnection < QP_PERCONNECTION_MIN) {
289 1 : HCCL_ERROR(
290 : "[TransportDeviceIbverbs][InitMemDetails]QPNum[%d] or qpInfos size[%u] is invalid",
291 : transDevIbverbsData_.qpsPerConnection, qpSize);
292 1 : return HCCL_E_INTERNAL;
293 : }
294 5 : combineAiQpInfo_.aiQpInfo.aiQpAddr = transDevIbverbsData_.qpInfo[0].qpPtr;
295 5 : combineAiQpInfo_.aiQpInfo.sqIndex = transDevIbverbsData_.qpInfo[0].sqIndex;
296 5 : combineAiQpInfo_.aiQpInfo.dbIndex = transDevIbverbsData_.qpInfo[0].dbIndex;
297 5 : combineAiQpInfos_.resize(transDevIbverbsData_.qpsPerConnection);
298 5 : for (u32 i = 1, j = 0; i < qpSize; i++, j++) {
299 0 : combineAiQpInfos_[j].aiQpInfo.aiQpAddr = transDevIbverbsData_.qpInfo[i].qpPtr;
300 0 : combineAiQpInfos_[j].aiQpInfo.sqIndex = transDevIbverbsData_.qpInfo[i].sqIndex;
301 0 : combineAiQpInfos_[j].aiQpInfo.dbIndex = transDevIbverbsData_.qpInfo[i].dbIndex;
302 : }
303 :
304 5 : CHK_RET(CheckDeviceId());
305 5 : CHK_RET(DlHnsFunction::GetInstance().DlHnsFunctionInit());
306 5 : transportAttr_.linkType = LinkType::LINK_ROCE;
307 5 : multiQpThreshold_ = transDevIbverbsData_.multiQpThreshold;
308 5 : qpsPerConnection_ = transDevIbverbsData_.qpsPerConnection;
309 5 : useAtomicWrite_ = transDevIbverbsData_.useAtomicWrite;
310 5 : HCCL_USER_CRITICAL_LOG(
311 : "create hccl transport:communicator[%s], local rank[%u], remote rank[%u],"
312 : "transporttype[%s], atomicWrite[%d]",
313 : machinePara_.tag.c_str(), machinePara_.localUserrank, machinePara_.remoteUserrank,
314 : GetLinkTypeEnumStr(GetLinkType()).c_str(), useAtomicWrite_);
315 5 : CHK_RET(BuildMemDetailsRmaMgrs());
316 5 : return HCCL_SUCCESS;
317 : }
318 :
319 5 : HcclResult TransportDeviceIbverbs::BuildMemDetailsRmaMgrs()
320 : {
321 5 : localMemDetailsRmaMgr_.reset();
322 5 : remoteMemDetailsRmaMgr_.reset();
323 5 : useMemDetailsLookup_ = false;
324 5 : localMemDetailsRmaMgr_ = std::make_unique<DeviceMemDetailsRmaMgr>();
325 5 : remoteMemDetailsRmaMgr_ = std::make_unique<DeviceMemDetailsRmaMgr>();
326 9 : for (const auto& md : transDevIbverbsData_.localRoceMemDetailsList) {
327 4 : if (md.size == 0U) {
328 0 : continue;
329 : }
330 4 : auto ent = std::make_shared<RoceMemDetails>(md);
331 4 : auto pr = localMemDetailsRmaMgr_->Add(MakeMemLookupKey(md.addr, md.size), ent);
332 4 : if (pr.first == localMemDetailsRmaMgr_->End()) {
333 0 : HCCL_ERROR(
334 : "[TransportDeviceIbverbs][BuildMemDetailsRmaMgrs] add local mem range failed, "
335 : "logical[0x%llx, +%llu) devBase[0x%llx] key[%u]",
336 : static_cast<unsigned long long>(md.addr), static_cast<unsigned long long>(md.size),
337 : static_cast<unsigned long long>(md.devAddr), md.key);
338 0 : return HCCL_E_INTERNAL;
339 : }
340 4 : HCCL_DEBUG(
341 : "[TransportDeviceIbverbs][BuildMemDetailsRmaMgrs] add local MR logical[0x%llx, +%llu) "
342 : "devBase[0x%llx] key[%u]",
343 : static_cast<unsigned long long>(md.addr), static_cast<unsigned long long>(md.size),
344 : static_cast<unsigned long long>(md.devAddr), md.key);
345 4 : }
346 9 : for (const auto& md : transDevIbverbsData_.remoteRoceMemDetailsList) {
347 4 : if (md.size == 0U) {
348 0 : continue;
349 : }
350 4 : auto ent = std::make_shared<RoceMemDetails>(md);
351 4 : auto pr = remoteMemDetailsRmaMgr_->Add(MakeMemLookupKey(md.addr, md.size), ent);
352 4 : if (pr.first == remoteMemDetailsRmaMgr_->End()) {
353 0 : HCCL_ERROR(
354 : "[TransportDeviceIbverbs][BuildMemDetailsRmaMgrs] add remote mem range failed, "
355 : "logical[0x%llx, +%llu) devBase[0x%llx] key[%u]",
356 : static_cast<unsigned long long>(md.addr), static_cast<unsigned long long>(md.size),
357 : static_cast<unsigned long long>(md.devAddr), md.key);
358 0 : return HCCL_E_INTERNAL;
359 : }
360 4 : HCCL_DEBUG(
361 : "[TransportDeviceIbverbs][BuildMemDetailsRmaMgrs] add remote MR logical[0x%llx, +%llu) "
362 : "devBase[0x%llx] key[%u]",
363 : static_cast<unsigned long long>(md.addr), static_cast<unsigned long long>(md.size),
364 : static_cast<unsigned long long>(md.devAddr), md.key);
365 4 : }
366 5 : HCCL_INFO(
367 : "[TransportDeviceIbverbs][BuildMemDetailsRmaMgrs] indexed localMR[%zu] remoteMR[%zu]",
368 : localMemDetailsRmaMgr_->size(), remoteMemDetailsRmaMgr_->size());
369 5 : useMemDetailsLookup_ = true;
370 5 : return HCCL_SUCCESS;
371 : }
372 :
373 0 : HcclResult TransportDeviceIbverbs::AddWrList(
374 : void* dstMemPtr, const void* srcMemPtr, u64 srcMemSize, u32 srcKey, u32 dstKey, WqeType wqeType, WrAuxInfo& aux,
375 : std::vector<WrInformation>& wrInfoVec)
376 : {
377 0 : HCCL_DEBUG("TransportDeviceIbverbs AddWrList start");
378 0 : if (srcMemSize == 0) {
379 0 : return HCCL_SUCCESS;
380 : }
381 0 : WrInformation wrInfoTmp;
382 0 : wrInfoTmp.wrData.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(dstMemPtr));
383 0 : wrInfoTmp.wrData.rkey = dstKey;
384 0 : wrInfoTmp.wrData.sendFlags = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
385 0 : fence_ = false;
386 0 : wrInfoTmp.wrData.immData = 0;
387 0 : wrInfoTmp.wrData.wrId = 0;
388 0 : wrInfoTmp.wrData.memList.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
389 0 : wrInfoTmp.wrData.memList.len = srcMemSize;
390 0 : wrInfoTmp.wrData.memList.lkey = srcKey;
391 :
392 0 : switch (wqeType) {
393 0 : case WqeType::WQE_TYPE_DATA:
394 : case WqeType::WQE_TYPE_DATA_NOTIFY:
395 : case WqeType::WQE_TYPE_ACK_NOTIFY:
396 : case WqeType::WQE_TYPE_DATA_ACK_NOTIFY:
397 : case WqeType::WQE_TYPE_DATA_WITH_NOTIFY:
398 0 : wrInfoTmp.wrData.op = RA_WR_RDMA_WRITE;
399 0 : wrInfoTmp.type = static_cast<u64>(wqeType);
400 0 : break;
401 0 : case WqeType::WQE_TYPE_DATA_WITH_REDUCE:
402 0 : wrInfoTmp.wrData.op = RA_WR_RDMA_REDUCE_WRITE;
403 0 : wrInfoTmp.wrData.aux = aux;
404 : // REDUCE WRITE 作为特殊的DATA
405 0 : wrInfoTmp.type = static_cast<u64>(WqeType::WQE_TYPE_DATA);
406 0 : break;
407 0 : case WqeType::WQE_TYPE_READ_DATA:
408 0 : wrInfoTmp.wrData.op = RA_WR_RDMA_READ;
409 0 : wrInfoTmp.type = static_cast<u64>(wqeType);
410 0 : break;
411 0 : default:
412 0 : HCCL_ERROR("error wqeType[%d]", wqeType);
413 0 : return HCCL_E_INTERNAL;
414 : }
415 0 : CHK_RET(GetWrDataAddr(dstMemPtr, wqeType, wrInfoTmp.wrDataAddr, wrInfoTmp.notifyId));
416 0 : HCCL_DEBUG(
417 : "wrInfoTmp dst_addr[0x%llx] memList addr[0x%llx] len[%llu]", wrInfoTmp.wrData.dstAddr,
418 : wrInfoTmp.wrData.memList.addr, srcMemSize);
419 0 : wrInfoVec.push_back(wrInfoTmp);
420 0 : HCCL_DEBUG("TransportDeviceIbverbs AddWrList end");
421 0 : return HCCL_SUCCESS;
422 : }
423 :
424 : HcclResult
425 0 : TransportDeviceIbverbs::GetMemInfo(UserMemType memType, void** dstMemPtr, unsigned int* dstKey, u64& dstMemSize)
426 : {
427 0 : CHK_PTR_NULL(dstMemPtr);
428 0 : CHK_PTR_NULL(dstKey);
429 :
430 0 : switch (memType) {
431 0 : case UserMemType::INPUT_MEM: {
432 0 : *dstMemPtr = remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].addr;
433 0 : dstMemSize = remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].len;
434 0 : *dstKey = remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].lkey;
435 0 : break;
436 : }
437 :
438 0 : case UserMemType::OUTPUT_MEM: {
439 0 : *dstMemPtr = remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].addr;
440 0 : dstMemSize = remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].len;
441 0 : *dstKey = remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].lkey;
442 0 : break;
443 : }
444 :
445 0 : default: {
446 0 : HCCL_ERROR("[Get][MemInfo]not support dst_mem_type=%d", memType);
447 0 : return HCCL_E_NOT_SUPPORT;
448 : }
449 : }
450 0 : return HCCL_SUCCESS;
451 : }
452 :
453 0 : HcclResult TransportDeviceIbverbs::ConstructPayLoadWqe(
454 : void* dstMemPtr, u32 dstKey, const void* src, u32 srcKey, u64 len, WqeType wqeType, WrAuxInfo& aux,
455 : std::vector<WrInformation>& wrInfoVec, u32 txSendDataTimes)
456 : {
457 : HcclResult ret;
458 : // 发送数据Wqe
459 0 : for (u32 txSendDataIdx = 0; txSendDataIdx < txSendDataTimes; txSendDataIdx++) {
460 0 : u64 txSendDataOffset = txSendDataIdx * RDMA_SEND_MAX_SIZE;
461 0 : u64 txSendDataSize = (txSendDataIdx == (txSendDataTimes - 1)) ? len - txSendDataOffset : RDMA_SEND_MAX_SIZE;
462 :
463 0 : void* txdstMemPtr = reinterpret_cast<void*>(reinterpret_cast<char*>(dstMemPtr) + txSendDataOffset);
464 :
465 0 : const void* txsrcMemPtr = reinterpret_cast<const void*>(reinterpret_cast<const char*>(src) + txSendDataOffset);
466 0 : ret = AddWrList(txdstMemPtr, txsrcMemPtr, txSendDataSize, srcKey, dstKey, wqeType, aux, wrInfoVec);
467 0 : CHK_PRT_RET(
468 : ret != HCCL_SUCCESS,
469 : HCCL_ERROR(
470 : "[TransportDeviceIbverbs][TxAsync]errNo[0x%016llx] In lbv exp, add wqe list failed."
471 : "srcMemSize[%llu]",
472 : HCCL_ERROR_CODE(ret), txSendDataSize),
473 : ret);
474 : }
475 0 : HCCL_DEBUG("TransportDeviceIbverbs TxPayLoad end");
476 :
477 0 : return HCCL_SUCCESS;
478 : }
479 :
480 0 : HcclResult TransportDeviceIbverbs::TxPayLoad(
481 : UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, WqeType wqeType, WrAuxInfo& aux,
482 : std::vector<WrInformation>& wrInfoVec)
483 : {
484 0 : HCCL_DEBUG("TransportDeviceIbverbs TxPayLoad start");
485 0 : void* dstMemPtr = nullptr;
486 : unsigned int dstKey;
487 : unsigned int srcKey;
488 0 : u64 dstMemSize = 0;
489 : // 为保证单算子下不同数据量下子图的结构相同，zero byte message 时也需要下发task
490 0 : u32 txSendDataTimes = (len + RDMA_SEND_MAX_SIZE - 1) / RDMA_SEND_MAX_SIZE;
491 :
492 : // 当前len不可用，无法校验dstOffset > dstMemSize
493 0 : CHK_RET(GetMemInfo(dstMemType, &dstMemPtr, &dstKey, dstMemSize));
494 :
495 0 : u64 srcAddr = reinterpret_cast<u64>(src);
496 0 : if (srcAddr >= localInputMem_.addr && srcAddr < localInputMem_.addr + localInputMem_.size) {
497 0 : srcKey = localInputMem_.key;
498 0 : } else if (srcAddr >= localOutputMem_.addr && srcAddr <= localOutputMem_.addr + localOutputMem_.size) {
499 0 : srcKey = localOutputMem_.key;
500 : } else {
501 0 : HCCL_ERROR(
502 : "[TransportDeviceIbverbs][TxAsync]src_ptr=%p is out of range, inputmem src[%p], size[%llu];"
503 : " outputmem src[%p] size[%llu]",
504 : src, localInputMem_.addr, localInputMem_.size, localOutputMem_.addr, localOutputMem_.size);
505 0 : return HCCL_E_INTERNAL;
506 : }
507 :
508 0 : dstMemPtr = reinterpret_cast<void*>(reinterpret_cast<char*>(dstMemPtr) + dstOffset);
509 0 : CHK_RET(ConstructPayLoadWqe(dstMemPtr, dstKey, src, srcKey, len, wqeType, aux, wrInfoVec, txSendDataTimes));
510 :
511 0 : return HCCL_SUCCESS;
512 : }
513 :
514 : HcclResult
515 0 : TransportDeviceIbverbs::TxAsync(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
516 : {
517 0 : CHK_SMART_PTR_NULL(stream);
518 0 : std::vector<WrInformation> wrInfoVec;
519 0 : struct WrAuxInfo aux = {};
520 0 : HCCL_DEBUG("TX src[%p] len[%llu] dstOffset[%llu]", src, len, dstOffset);
521 :
522 0 : if (len > 0) {
523 0 : CHK_PTR_NULL(src);
524 0 : CHK_RET(TxPayLoad(dstMemType, dstOffset, src, len, WqeType::WQE_TYPE_DATA, aux, wrInfoVec));
525 : }
526 :
527 0 : CHK_RET(TxSendDataAndNotify(wrInfoVec, stream, GetUseOneDoorbellValue()));
528 0 : return HCCL_SUCCESS;
529 0 : }
530 :
531 0 : HcclResult TransportDeviceIbverbs::TxWithReduce(
532 : UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, const HcclDataType datatype, HcclReduceOp redOp,
533 : Stream& stream)
534 : {
535 0 : CHK_SMART_PTR_NULL(stream);
536 0 : std::vector<WrInformation> wrInfoVec;
537 0 : struct WrAuxInfo aux = {};
538 0 : aux.dataType = RDMA_REDUCE_DATA_TYPE_TABLE[datatype];
539 0 : aux.reduceType = RDMA_REDUCE_OP_TYPE_TABLE[redOp];
540 0 : if (aux.dataType == static_cast<uint8_t>(RdmaReduceDataType::RDMA_REDUCE_DATA_INVALID)
541 0 : || aux.reduceType == static_cast<uint8_t>(RdmaReduceOpType::RDMA_REDUCE_OP_INVALID)) {
542 0 : HCCL_ERROR(
543 : "unsupported data type [%s] or Reduce type [%s]", GetDataTypeEnumStr(datatype).c_str(),
544 : GetReduceOpEnumStr(redOp).c_str());
545 0 : return HCCL_E_INTERNAL;
546 : }
547 0 : if (len > 0) {
548 0 : CHK_PTR_NULL(src);
549 0 : CHK_RET(TxPayLoad(dstMemType, dstOffset, src, len, WqeType::WQE_TYPE_DATA_WITH_REDUCE, aux, wrInfoVec));
550 : }
551 :
552 0 : CHK_RET(TxSendDataAndNotify(wrInfoVec, stream, GetUseOneDoorbellValue()));
553 0 : return HCCL_SUCCESS;
554 0 : }
555 :
556 0 : HcclResult TransportDeviceIbverbs::TxWithReduce(
557 : const std::vector<TxMemoryInfo>& txWithReduceMems, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream)
558 : {
559 0 : CHK_SMART_PTR_NULL(stream);
560 0 : std::vector<WrInformation> wrInfoVec;
561 0 : struct WrAuxInfo aux = {};
562 0 : aux.dataType = RDMA_REDUCE_DATA_TYPE_TABLE[datatype];
563 0 : aux.reduceType = RDMA_REDUCE_OP_TYPE_TABLE[redOp];
564 0 : if (aux.dataType == static_cast<uint8_t>(RdmaReduceDataType::RDMA_REDUCE_DATA_INVALID)
565 0 : || aux.reduceType == static_cast<uint8_t>(RdmaReduceOpType::RDMA_REDUCE_OP_INVALID)) {
566 0 : HCCL_ERROR(
567 : "unsupported data type [%s] or Reduce type [%s]", GetDataTypeEnumStr(datatype).c_str(),
568 : GetReduceOpEnumStr(redOp).c_str());
569 0 : return HCCL_E_INTERNAL;
570 : }
571 :
572 0 : for (const TxMemoryInfo& txWithReduceMem : txWithReduceMems) {
573 0 : if (txWithReduceMem.len == 0) {
574 0 : continue;
575 : }
576 0 : CHK_PTR_NULL(txWithReduceMem.src);
577 0 : CHK_RET(TxPayLoad(
578 : txWithReduceMem.dstMemType, txWithReduceMem.dstOffset, txWithReduceMem.src, txWithReduceMem.len,
579 : WqeType::WQE_TYPE_DATA_WITH_REDUCE, aux, wrInfoVec));
580 : }
581 :
582 0 : CHK_RET(TxSendDataAndNotify(wrInfoVec, stream, GetUseOneDoorbellValue()));
583 0 : return HCCL_SUCCESS;
584 0 : }
585 :
586 0 : HcclResult TransportDeviceIbverbs::TxSendDataAndNotifyWithSingleQP(
587 : std::vector<WrInformation>& wrInfoVec, Stream& stream, bool useOneDoorbell)
588 : {
589 : // 发送data notify同步信息
590 0 : struct WrAuxInfo aux = {};
591 0 : void* remoteNotifyaddr = remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].addr;
592 : ;
593 0 : CHK_RET(AddWrList(
594 : remoteNotifyaddr, notifyValueAddr_, notifySize_, memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey,
595 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].lkey, WqeType::WQE_TYPE_DATA_NOTIFY, aux, wrInfoVec));
596 :
597 0 : CHK_RET(RdmaSendAsync(wrInfoVec, stream, useOneDoorbell));
598 0 : return HCCL_SUCCESS;
599 : }
600 :
601 : HcclResult
602 0 : TransportDeviceIbverbs::TxSendDataAndNotify(std::vector<WrInformation>& wrInfoVec, Stream& stream, bool useOneDoorbell)
603 : {
604 0 : u32 maxLength = 0;
605 0 : for (u32 i = 0; i < wrInfoVec.size(); i++) {
606 0 : if (wrInfoVec[i].wrData.memList.len > maxLength) {
607 0 : maxLength = wrInfoVec[i].wrData.memList.len;
608 : }
609 : }
610 0 : u32 actualMultiQpNum = GetActualQpNum(maxLength);
611 0 : HCCL_DEBUG(
612 : "[TransportDeviceIbverbs][TxSendDataAndNotify] UseMultiQp[%d] MultiQpNum[%u] actualMultiQpNum[%u] "
613 : "maxLength[%u]",
614 : UseMultiQp(), qpsPerConnection_, actualMultiQpNum, maxLength);
615 0 : if (UseMultiQp() && actualMultiQpNum != 1 && actualMultiQpNum <= qpsPerConnection_ && maxLength != 0) {
616 0 : CHK_RET(TxSendDataAndNotifyWithMultiQP(wrInfoVec, actualMultiQpNum, stream, useOneDoorbell));
617 : } else {
618 0 : CHK_RET(TxSendDataAndNotifyWithSingleQP(wrInfoVec, stream, useOneDoorbell));
619 : }
620 0 : return HCCL_SUCCESS;
621 : }
622 :
623 0 : HcclResult TransportDeviceIbverbs::TxAsync(std::vector<TxMemoryInfo>& txMems, Stream& stream)
624 : {
625 0 : CHK_SMART_PTR_NULL(stream);
626 :
627 0 : std::vector<WrInformation> wrInfoVec;
628 0 : struct WrAuxInfo aux = {};
629 :
630 0 : for (auto& mem : txMems) {
631 0 : HCCL_DEBUG("TX src[%p] len[%llu] dstOffset[%llu]", mem.src, mem.len, mem.dstOffset);
632 0 : if (mem.len == 0) {
633 0 : continue;
634 : }
635 0 : CHK_PTR_NULL(mem.src);
636 0 : CHK_RET(TxPayLoad(mem.dstMemType, mem.dstOffset, mem.src, mem.len, WqeType::WQE_TYPE_DATA, aux, wrInfoVec));
637 : }
638 :
639 0 : CHK_RET(TxSendDataAndNotify(wrInfoVec, stream, GetUseOneDoorbellValue()));
640 0 : return HCCL_SUCCESS;
641 0 : }
642 :
643 0 : HcclResult TransportDeviceIbverbs::TxWrList(
644 : std::vector<WrInformation>& wrInfoVec, Stream& stream, std::vector<struct SendWrRsp>& opRspVec, u32 multiQpIndex)
645 : {
646 : (void)stream;
647 :
648 0 : u32 totalWqeCount = wrInfoVec.size();
649 0 : WrInformation* wrlist = wrInfoVec.data();
650 0 : struct SendWrRsp* opRsp = opRspVec.data();
651 :
652 : // HCCP会校验 zero byte messages 的内存地址是否已注册MR。对于 zero byte messages 不下发WR，将opRsp设置为特殊值。
653 : // 下发rdmasend task时检查该特殊值，如果zero byte message则不下发rdmasend task。
654 0 : bool batchSendWr = true;
655 0 : for (u32 i = 0; i < totalWqeCount; i++) {
656 0 : if (wrInfoVec[i].wrData.memList.len == 0) {
657 0 : batchSendWr = false;
658 0 : break;
659 : }
660 : }
661 :
662 0 : if (batchSendWr) {
663 0 : CHK_RET(SendWrList(totalWqeCount, wrlist, opRsp, multiQpIndex));
664 : } else {
665 0 : for (u32 i = 0; i < totalWqeCount; i++) {
666 0 : if (wrInfoVec[i].wrData.memList.len > 0) {
667 0 : CHK_RET(SendWrList(1U, &wrlist[i], &opRsp[i], multiQpIndex));
668 : } else {
669 0 : opRsp[i].wqeTmp.sqIndex = INVALID_UINT;
670 0 : opRsp[i].wqeTmp.wqeIndex = INVALID_UINT;
671 0 : opRsp[i].db.dbIndex = INVALID_UINT;
672 0 : opRsp[i].db.dbInfo = INVALID_U64;
673 : }
674 : }
675 : }
676 :
677 0 : return HCCL_SUCCESS;
678 : }
679 :
680 : HcclResult
681 1 : TransportDeviceIbverbs::SendWrList(u32 wrNum, WrInformation* wrlist, struct SendWrRsp* opRsp, u32 multiQpIndex)
682 : {
683 1 : unsigned int completeNum = 0;
684 1 : HcclResult ret = SendWrlistExt(wrlist, opRsp, wrNum, &completeNum, multiQpIndex);
685 1 : CHK_PRT_RET(
686 : ret != HCCL_SUCCESS,
687 : HCCL_ERROR("[TransportDeviceIbverbs][SendWrList]In ibv send wq list, SendWrlistExt failed.ret[%d]", ret),
688 : HCCL_E_NETWORK);
689 1 : return HCCL_SUCCESS;
690 : }
691 :
692 0 : HcclResult TransportDeviceIbverbs::SendWrlistExt(
693 : WrInformation wr[], struct SendWrRsp opRsp[], unsigned int sendNum, unsigned int* completeNum, u32 multiQpIndex)
694 : {
695 0 : HcclResult ret = HCCL_SUCCESS;
696 0 : auto startTime = std::chrono::steady_clock::now();
697 0 : u32 remainNum = sendNum;
698 0 : unsigned int completeNumLocal = 0;
699 0 : *completeNum = 0;
700 : while (true) {
701 0 : if (remainNum > sendNum) {
702 0 : HCCL_ERROR(
703 : "[Aicpu][Send][Wr]wr list send async fail. return[%d], remainNum[%u], "
704 : "sendNum[%u].",
705 : HCCL_E_ROCE_TRANSFER, remainNum, sendNum);
706 0 : return HCCL_E_ROCE_TRANSFER;
707 : }
708 0 : if (remainNum == 0) {
709 0 : break;
710 : }
711 0 : ret = TxSendWrlistExt(
712 0 : wr + (sendNum - remainNum), remainNum, opRsp + (sendNum - remainNum), &completeNumLocal, multiQpIndex);
713 0 : *completeNum += completeNumLocal;
714 0 : if (ret == HCCL_SUCCESS && *completeNum == sendNum) {
715 0 : break; // 成功跳出
716 : }
717 :
718 0 : if (ret == HCCL_E_AGAIN || *completeNum < sendNum) {
719 0 : remainNum -= completeNumLocal;
720 0 : bool bTimeout = ((std::chrono::steady_clock::now() - startTime) >= timeout_);
721 0 : CHK_PRT_RET(
722 : bTimeout,
723 : HCCL_ERROR(
724 : "[Aicpu][Send][Wr]errNo[0x%016llx] wrlist send async timeout[%d]ms. "
725 : "return[%d], params: send_wrAddr[%p], opRspAddr[%p]",
726 : HCCL_ERROR_CODE(HCCL_E_ROCE_TRANSFER), timeout_, ret, wr, opRsp),
727 : HCCL_E_ROCE_TRANSFER);
728 0 : SaluSleep(ONE_MILLISECOND_OF_USLEEP);
729 0 : } else {
730 0 : HCCL_ERROR(
731 : "[Aicpu][Send][Wr]wrlist send async fail. return[%d], para: send_wrAddr[%p], "
732 : "opRspAddr[%p].",
733 : ret, wr, opRsp);
734 0 : return HCCL_E_ROCE_TRANSFER; // 非-2/-11场景错误，不轮询，直接退出
735 : }
736 0 : }
737 0 : return HCCL_SUCCESS;
738 : }
739 :
740 0 : HcclResult TransportDeviceIbverbs::TxSendWrlistExt(
741 : WrInformation wrList[], u32 sendNum, struct SendWrRsp opRsp[], unsigned int* completeNum, u32 multiQpIndex)
742 : {
743 0 : u32 i = 0;
744 0 : s32 ret = 0;
745 0 : struct ibv_send_wr ib_wr = {};
746 0 : struct ibv_sge list = {};
747 0 : struct ibv_send_wr* bad_wr = nullptr;
748 0 : struct WrExpRsp exp_rsp = {};
749 0 : struct IbvPostSendExtResp ext_rsp = {};
750 0 : struct IbvPostSendExtAddt ext_attr = {};
751 0 : for (; i < sendNum; i++) {
752 0 : if (wrList[i].wrData.memList.len > IBV_SGLIST_LEN_MAX) {
753 0 : HCCL_ERROR("sg list len is more than 2G, len[%u]", wrList[i].wrData.memList.len);
754 0 : return HCCL_E_PARA;
755 : }
756 0 : u64 wrIdoffset = wrIdOffset_++;
757 :
758 : // 910B和910_93，reduce的下一个notify要设置为atomic write
759 : u32& preWrOpcode = multiQpIndex == RDMA_INVALID_QP_INDEX ? combineAiQpInfo_.preWrOpcode :
760 0 : combineAiQpInfos_[multiQpIndex].preWrOpcode;
761 0 : ModifyAtomicWriteAfterReduce(preWrOpcode, wrList[i].type, wrList[i].wrData.op, wrList[i].wrData.immData);
762 :
763 0 : if (wrList[i].wrData.op != RA_WR_SEND && wrList[i].wrData.op != RA_WR_SEND_WITH_IMM) {
764 0 : HCCL_DEBUG("remote wr dst addr is 0x%llx", wrList[i].wrData.dstAddr);
765 0 : list.addr = wrList[i].wrData.memList.addr;
766 0 : list.length = wrList[i].wrData.memList.len;
767 0 : list.lkey = wrList[i].wrData.memList.lkey;
768 :
769 0 : ib_wr.sg_list = &list;
770 0 : ib_wr.opcode = static_cast<enum ibv_wr_opcode>(wrList[i].wrData.op);
771 0 : ib_wr.send_flags = static_cast<unsigned int>(wrList[i].wrData.sendFlags);
772 0 : ib_wr.imm_data = wrList[i].wrData.immData;
773 :
774 0 : ib_wr.num_sge = 1; /* only support one sge */
775 0 : ib_wr.wr_id = wrList[i].wrData.wrId += wrIdoffset;
776 0 : ib_wr.wr.rdma.rkey = wrList[i].wrData.rkey;
777 0 : ib_wr.wr.rdma.remote_addr = wrList[i].wrData.dstAddr;
778 : } else {
779 0 : list.addr = wrList[i].wrData.memList.addr;
780 0 : list.length = wrList[i].wrData.memList.len;
781 0 : list.lkey = wrList[i].wrData.memList.lkey;
782 :
783 0 : ib_wr.sg_list = &list;
784 0 : ib_wr.opcode = static_cast<enum ibv_wr_opcode>(wrList[i].wrData.op);
785 0 : ib_wr.send_flags = static_cast<unsigned int>(wrList[i].wrData.sendFlags);
786 0 : ib_wr.imm_data = wrList[i].wrData.immData;
787 :
788 0 : ib_wr.num_sge = 1; /* only support one sge */
789 0 : ib_wr.wr_id = wrList[i].wrData.wrId += wrIdoffset;
790 : }
791 0 : unsigned long long aiQpAddr = multiQpIndex == RDMA_INVALID_QP_INDEX ?
792 : combineAiQpInfo_.aiQpInfo.aiQpAddr :
793 0 : combineAiQpInfos_[multiQpIndex].aiQpInfo.aiQpAddr;
794 0 : struct ibv_qp* qp = reinterpret_cast<struct ibv_qp*>(aiQpAddr);
795 0 : HCCL_DEBUG(
796 : "ib_wr.sglist[%u].addr[%p], ib_wr.sglist[%u].length[%u], ib_wr.sglist[%u], "
797 : "ib_wr.wr_id[%llu], raddr[%p], opcode[%d], imm_data[0x%llx]",
798 : i, list.addr, i, list.length, i, ib_wr.wr_id, ib_wr.wr.rdma.remote_addr, wrList[i].wrData.op,
799 : ib_wr.imm_data);
800 0 : if (wrList[i].wrData.op == RA_WR_RDMA_ATOMIC_WRITE) {
801 0 : ext_attr.reduce_op = wrList[i].wrData.aux.reduceType;
802 0 : ext_attr.reduce_type = wrList[i].wrData.aux.dataType;
803 0 : ret = DlHnsFunction::GetInstance().dlHnsIbvExtPostSend(qp, &ib_wr, &bad_wr, &ext_attr, &ext_rsp);
804 0 : HCOMM_DSB();
805 0 : exp_rsp.wqe_index = ext_rsp.wqe_index;
806 0 : exp_rsp.db_info = ext_rsp.db_info;
807 0 : HCCL_DEBUG(
808 : "ibv_ext_post_send, op = [0x%x], imm_data = [0x%lx], reduce_op = [%d], reduceType = [%d]",
809 : wrList[i].wrData.op, ib_wr.imm_data, ext_attr.reduce_op, ext_attr.reduce_type);
810 0 : } else if (
811 0 : wrList[i].wrData.op == RA_WR_RDMA_WRITE_WITH_NOTIFY || wrList[i].wrData.op == RA_WR_RDMA_REDUCE_WRITE
812 0 : || wrList[i].wrData.op == RA_WR_RDMA_REDUCE_WRITE_WITH_NOTIFY) {
813 0 : ib_wr.imm_data = htobe32(
814 : (wrList[i].wrData.aux.notifyOffset & RDMA_WRITE_NOTIFY_OFFSET_MASK) | RDMA_WRITE_NOTIFY_VALUE_RECORD);
815 0 : ext_attr.reduce_op = wrList[i].wrData.aux.reduceType;
816 0 : ext_attr.reduce_type = wrList[i].wrData.aux.dataType;
817 0 : ret = DlHnsFunction::GetInstance().dlHnsIbvExtPostSend(qp, &ib_wr, &bad_wr, &ext_attr, &ext_rsp);
818 0 : HCOMM_DSB();
819 0 : exp_rsp.wqe_index = ext_rsp.wqe_index;
820 0 : exp_rsp.db_info = ext_rsp.db_info;
821 0 : HCCL_DEBUG(
822 : "ibv_ext_post_send, op = [0x%x], imm_data = [0x%lx], reduce_op = [%d],reduceType = [%d]",
823 : wrList[i].wrData.op, ib_wr.imm_data, ext_attr.reduce_op, ext_attr.reduce_type);
824 : } else {
825 0 : ret = DlHnsFunction::GetInstance().dlHnsIbvExpPostSend(qp, &ib_wr, &bad_wr, &exp_rsp);
826 0 : HCOMM_DSB();
827 0 : HCCL_DEBUG(
828 : "ibv_exp_post_send, op = [0x%x], remote_addr = [0x%llx], size = [%d]", wrList[i].wrData.op,
829 : ib_wr.wr.rdma.remote_addr, ib_wr.sg_list->length);
830 : }
831 0 : if (ret) {
832 0 : HCCL_WARNING("[TxSendWrlistExt]ibv_post_send failed ret %d, i[%u]", ret, i);
833 0 : break;
834 : }
835 0 : unsigned long long dbIndex = multiQpIndex == RDMA_INVALID_QP_INDEX ?
836 0 : combineAiQpInfo_.aiQpInfo.dbIndex :
837 0 : combineAiQpInfos_[multiQpIndex].aiQpInfo.dbIndex;
838 0 : opRsp[i].db.dbIndex = (unsigned int)dbIndex;
839 0 : HCCL_DEBUG("opRsp.db.dbIndex = [%d]", opRsp[i].db.dbIndex);
840 0 : opRsp[i].db.dbInfo = exp_rsp.db_info;
841 : }
842 :
843 0 : HCCL_DEBUG("completeNum[%d], ret[%d]", i, ret);
844 0 : *completeNum = i;
845 0 : if ((ret == SOCK_ENOENT) || (ret == ROCE_EAGAIN)
846 0 : || (workFlowMode_ == HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE && ret == ROCE_ENOMEM)) {
847 0 : return HCCL_E_AGAIN;
848 0 : } else if (!ret) {
849 0 : return HCCL_SUCCESS;
850 0 : } else if (ret == RDMA_QP_NO_MEM) { // 表示qp已满，内存不足，需要重发
851 0 : ib_wr.wr_id = wrList[i].wrData.wrId -= wrIdOffset_;
852 : // 可能出现主流没有launch，但从流一直下发导致卡死超时的问题，所以这里将所有流都下发
853 0 : CHK_RET(dispatcher_->LaunchAllTasks());
854 0 : return HCCL_E_AGAIN;
855 : } else {
856 0 : return HCCL_E_ROCE_TRANSFER;
857 : }
858 :
859 : return HCCL_SUCCESS;
860 : }
861 :
862 0 : HcclResult TransportDeviceIbverbs::RdmaSendAsync(
863 : std::vector<WrInformation>& wrInfoVec, Stream& stream, bool useOneDoorbell, u32 multiQpIndex)
864 : {
865 : HcclResult ret;
866 :
867 0 : std::vector<struct SendWrRsp> opRspVec(wrInfoVec.size());
868 0 : CHK_RET(TxWrList(wrInfoVec, stream, opRspVec, multiQpIndex));
869 :
870 0 : for (u32 i = 0; i < wrInfoVec.size(); i++) {
871 0 : if (useOneDoorbell && i != wrInfoVec.size() - 1) {
872 : // 如果useOneDoorbell为true，只敲最后一次doorbell
873 0 : continue;
874 : }
875 :
876 0 : RdmaTaskInfo taskInfo = {};
877 0 : taskInfo.remoteRank = machinePara_.remoteWorldRank;
878 0 : taskInfo.rdmaType = (wrInfoVec[i].type == static_cast<u64>(WqeType::WQE_TYPE_DATA)) ?
879 : RdmaType::RDMA_SEND_PAYLOAD :
880 : RdmaType::RDMA_SEND_NOTIFY;
881 :
882 0 : if (useOneDoorbell) {
883 : // 如果useOneDoorbell为true，一次性传入所有wr
884 0 : taskInfo.wrInfos = wrInfoVec;
885 : } else {
886 0 : taskInfo.wrInfos.push_back(wrInfoVec[i]);
887 : }
888 :
889 0 : u32 dbIndex = static_cast<u32>(opRspVec[i].db.dbIndex);
890 0 : HCCL_DEBUG("dbIndex = [%d]", dbIndex);
891 0 : u64 dbInfo = static_cast<u64>(opRspVec[i].db.dbInfo);
892 :
893 0 : ret = dispatcher_->RdmaSend(dbIndex, dbInfo, stream, taskInfo);
894 0 : CHK_PRT_RET(
895 : ret != HCCL_SUCCESS,
896 : HCCL_ERROR(
897 : "[TransportDeviceIbverbs][RdmaSendAsync]errNo[0x%016llx] In lbv exp op base mode, "
898 : "rdma send failed. dbIndex[%u] dbInfo[%llu] wqe type[%llu] addr[%llu]",
899 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo, wrInfoVec[i].type, wrInfoVec[i].wrDataAddr),
900 : ret);
901 0 : }
902 0 : return HCCL_SUCCESS;
903 0 : }
904 :
905 : HcclResult
906 1 : TransportDeviceIbverbs::RdmaSendAsync(struct SendWr& wr, Stream& stream, WqeType wqeType, u64 notifyAddr, u32 notifyId)
907 : {
908 : HcclResult ret;
909 1 : WrInformation wrInfoTmp;
910 1 : struct SendWrRsp opRsp = {};
911 1 : struct WrAuxInfo aux = {};
912 1 : wrInfoTmp.wrData.memList = wr.bufList[0];
913 1 : wrInfoTmp.wrData.dstAddr = wr.dstAddr;
914 1 : wrInfoTmp.wrData.op = wr.op;
915 1 : wrInfoTmp.wrData.sendFlags = wr.sendFlag;
916 1 : wrInfoTmp.wrData.immData = 0;
917 1 : wrInfoTmp.wrData.wrId = 0xFF;
918 1 : wrInfoTmp.wrData.rkey = wr.rkey;
919 1 : wrInfoTmp.wrData.aux = aux;
920 :
921 1 : CHK_RET(SendWrList(1U, &wrInfoTmp, &opRsp));
922 1 : u32 dbIndex = static_cast<u32>(opRsp.db.dbIndex);
923 1 : u64 dbInfo = static_cast<u64>(opRsp.db.dbInfo);
924 1 : HCCL_DEBUG("dbIndex = [%d]", dbIndex);
925 1 : RdmaTaskInfo taskInfo = {};
926 1 : taskInfo.remoteRank = machinePara_.remoteWorldRank;
927 1 : taskInfo.rdmaType = (wqeType == WqeType::WQE_TYPE_DATA) ? RdmaType::RDMA_SEND_PAYLOAD : RdmaType::RDMA_SEND_NOTIFY;
928 1 : wrInfoTmp.type = static_cast<u64>(wqeType);
929 1 : wrInfoTmp.wrDataAddr = notifyAddr;
930 1 : wrInfoTmp.notifyId = notifyId;
931 1 : taskInfo.wrInfos.push_back(wrInfoTmp);
932 :
933 1 : ret = dispatcher_->RdmaSend(dbIndex, dbInfo, stream, taskInfo);
934 1 : CHK_PRT_RET(
935 : ret != HCCL_SUCCESS,
936 : HCCL_ERROR(
937 : "[TransportDeviceIbverbs][RdmaSendAsync]errNo[0x%016llx] In lbv exp op base mode, "
938 : "rdma send failed. dbIndex[%u] dbInfo[%llu], addr[%llu]",
939 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo, notifyAddr),
940 : ret);
941 1 : return HCCL_SUCCESS;
942 1 : }
943 :
944 0 : HcclResult TransportDeviceIbverbs::GetWrDataAddr(void* dstAddr, WqeType wqeType, u64& wrDataAddr, u32& notifyId)
945 : {
946 0 : switch (wqeType) {
947 0 : case WqeType::WQE_TYPE_DATA:
948 : case WqeType::WQE_TYPE_DATA_WITH_NOTIFY:
949 : case WqeType::WQE_TYPE_DATA_WITH_REDUCE:
950 : case WqeType::WQE_TYPE_READ_DATA:
951 0 : wrDataAddr = reinterpret_cast<u64>(dstAddr);
952 0 : notifyId = INVALID_UINT;
953 0 : break;
954 0 : case WqeType::WQE_TYPE_DATA_NOTIFY:
955 0 : wrDataAddr = reinterpret_cast<u64>(remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].addr);
956 0 : notifyId = remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].notifyId;
957 0 : break;
958 0 : case WqeType::WQE_TYPE_ACK_NOTIFY:
959 0 : wrDataAddr = reinterpret_cast<u64>(remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr);
960 0 : notifyId = remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].notifyId;
961 0 : break;
962 0 : case WqeType::WQE_TYPE_DATA_ACK_NOTIFY:
963 0 : wrDataAddr = reinterpret_cast<u64>(remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].addr);
964 0 : notifyId = remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].notifyId;
965 0 : break;
966 0 : default:
967 0 : HCCL_ERROR("[Get][WrDataAddr]error wqeType[%d]", wqeType);
968 0 : return HCCL_E_INTERNAL;
969 : }
970 0 : HCCL_DEBUG(
971 : "%s dstAddr:%p, wqeType:%d, wrDataAddr:%llu, notifyId:%u", __func__, dstAddr, wqeType, wrDataAddr, notifyId);
972 0 : return HCCL_SUCCESS;
973 : }
974 :
975 0 : HcclResult TransportDeviceIbverbs::TxSendWqe(
976 : void* dstMemPtr, u32 dstKey, const void* srcMemPtr, u32 srcKey, u64 srcMemSize, Stream& stream, WqeType wqeType)
977 : {
978 0 : struct SgList list = {};
979 0 : struct SendWr wr = {};
980 : // 构造wr信息
981 0 : list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
982 0 : list.len = srcMemSize;
983 0 : list.lkey = srcKey;
984 :
985 0 : wr.bufList = &list;
986 0 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
987 0 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(dstMemPtr));
988 0 : wr.rkey = dstKey;
989 0 : wr.op = 0; /* RDMA_WRITE: 0 */
990 0 : wr.sendFlag = fence_ ? (RA_SEND_SIGNALED | RA_SEND_FENCE) : RA_SEND_SIGNALED;
991 0 : fence_ = false;
992 :
993 : // 获取notify偏移地址，对于发送数据时，偏移地址为0
994 0 : u64 wrDataAddr = 0;
995 0 : u32 notifyId = INVALID_UINT;
996 0 : CHK_RET(GetWrDataAddr(dstMemPtr, wqeType, wrDataAddr, notifyId));
997 :
998 : // RDMA异步发送
999 0 : CHK_RET(RdmaSendAsync(wr, stream, wqeType, wrDataAddr, notifyId));
1000 0 : return HCCL_SUCCESS;
1001 : }
1002 :
1003 0 : HcclResult TransportDeviceIbverbs::RxAsync(
1004 : [[maybe_unused]] UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream)
1005 : {
1006 0 : CHK_SMART_PTR_NULL(stream);
1007 : // 等待TS把任务处理完成
1008 0 : HCCL_DEBUG("RX dst[%p] len[%llu] srcOffset[%llu]", dst, len, srcOffset);
1009 0 : u32 actualMultiQpNum = 1;
1010 0 : const u32 KByteToByte = 1024; // 1024 多QP阈值单位是KB
1011 0 : if (len / qpsPerConnection_ > multiQpThreshold_ * KByteToByte) {
1012 0 : actualMultiQpNum = qpsPerConnection_;
1013 : } else {
1014 0 : u32 quotient = len / (multiQpThreshold_ * KByteToByte);
1015 0 : u32 remainder = len % (multiQpThreshold_ * KByteToByte);
1016 0 : actualMultiQpNum = quotient + (remainder != 0 ? 1 : 0);
1017 : }
1018 0 : if (UseMultiQp() && actualMultiQpNum != 1 && actualMultiQpNum <= qpsPerConnection_ && len != 0) {
1019 0 : for (u32 i = 0; i < actualMultiQpNum; i++) {
1020 0 : CHK_RET(dispatcher_->SignalWait(
1021 : multiQpDataNotify_[i]->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank,
1022 : INVALID_VALUE_STAGE, false, multiQpDataNotify_[i]->notifyId_));
1023 : }
1024 : } else {
1025 0 : CHK_RET(dispatcher_->SignalWait(
1026 : dataNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE,
1027 : false, dataNotify_->notifyId_));
1028 : }
1029 0 : return HCCL_SUCCESS;
1030 : }
1031 :
1032 0 : HcclResult TransportDeviceIbverbs::RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream& stream)
1033 : {
1034 0 : CHK_PRT_RET(rxMems.size() == 0, HCCL_ERROR("Invalid rxMem size[%u]", rxMems.size()), HCCL_E_PARA);
1035 0 : CHK_SMART_PTR_NULL(stream);
1036 0 : for (auto& mem : rxMems) {
1037 0 : HCCL_DEBUG("RX dst[%p] len[%llu] dstOffset[%llu]", mem.dst, mem.len, mem.srcOffset);
1038 : }
1039 0 : u32 maxLength = 0;
1040 0 : for (u32 i = 0; i < rxMems.size(); i++) {
1041 0 : if (rxMems[i].len > maxLength) {
1042 0 : maxLength = rxMems[i].len;
1043 : }
1044 : }
1045 :
1046 0 : CHK_RET(RxAsync(rxMems[0].srcMemType, rxMems[0].srcOffset, rxMems[0].dst, maxLength, stream));
1047 0 : return HCCL_SUCCESS;
1048 : }
1049 :
1050 0 : HcclResult TransportDeviceIbverbs::DataReceivedAck(Stream& stream)
1051 : {
1052 0 : CHK_RET(PostFinAck(stream));
1053 0 : CHK_RET(WaitFinAck(stream));
1054 :
1055 0 : return HCCL_SUCCESS;
1056 : }
1057 :
1058 0 : HcclResult TransportDeviceIbverbs::TxWaitDone([[maybe_unused]] Stream& stream) { return HCCL_SUCCESS; }
1059 :
1060 : /* 发送ack消息(同步模式) */
1061 0 : HcclResult TransportDeviceIbverbs::TxAck(Stream& stream)
1062 : {
1063 0 : CHK_RET(TxSendWqe(
1064 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr,
1065 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].lkey,
1066 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1067 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream, WqeType::WQE_TYPE_ACK_NOTIFY));
1068 0 : return HCCL_SUCCESS;
1069 : }
1070 :
1071 : /* 接收ack消息(同步模式) */
1072 0 : HcclResult TransportDeviceIbverbs::RxAck(Stream& stream)
1073 : {
1074 0 : CHK_RET(dispatcher_->SignalWait(
1075 : ackNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE, false,
1076 : ackNotify_->notifyId_));
1077 0 : return HCCL_SUCCESS;
1078 : }
1079 :
1080 0 : HcclResult TransportDeviceIbverbs::TxDataSignal(Stream& stream)
1081 : {
1082 : // 发送data notify同步信息
1083 0 : void* remoteNotifyaddr = remoteDataNotifyMsg_.addr;
1084 0 : HcclResult ret = TxSendWqe(
1085 0 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].addr,
1086 0 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].lkey,
1087 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1088 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream, WqeType::WQE_TYPE_DATA_NOTIFY);
1089 :
1090 0 : CHK_PRT_RET(
1091 : ret != HCCL_SUCCESS,
1092 : HCCL_ERROR(
1093 : "[TransportDeviceIbverbs][TxDataSignal]errNo[0x%016llx] In ibv tx data signal, send notify "
1094 : "wqe failed. dstMemPtr[%p], srcMemPtr[%p], srcMemSize[%llu]",
1095 : HCCL_ERROR_CODE(ret), remoteNotifyaddr, notifyValueAddr_, notifySize_),
1096 : ret);
1097 : // 每发送一个data notify wqe, count 自增
1098 0 : return HCCL_SUCCESS;
1099 : }
1100 :
1101 0 : HcclResult TransportDeviceIbverbs::RxDataSignal(Stream& stream)
1102 : {
1103 : /* 等待send_ready_event事件 */
1104 0 : CHK_RET(dispatcher_->SignalWait(
1105 : dataNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE,
1106 : false, dataNotify_->notifyId_));
1107 0 : return HCCL_SUCCESS;
1108 : }
1109 :
1110 : /* 发送ack消息(同步模式) */
1111 0 : HcclResult TransportDeviceIbverbs::TxPrepare(Stream& stream)
1112 : {
1113 0 : CHK_RET(dispatcher_->SignalWait(
1114 : ackNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE, false,
1115 : ackNotify_->notifyId_));
1116 0 : return HCCL_SUCCESS;
1117 : }
1118 :
1119 : /* 接收ack消息(同步模式) */
1120 0 : HcclResult TransportDeviceIbverbs::RxPrepare(Stream& stream)
1121 : {
1122 0 : CHK_RET(TxSendWqe(
1123 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr,
1124 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].lkey,
1125 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1126 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream, WqeType::WQE_TYPE_ACK_NOTIFY));
1127 0 : return HCCL_SUCCESS;
1128 : }
1129 :
1130 : HcclResult
1131 0 : TransportDeviceIbverbs::TxData(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
1132 : {
1133 0 : CHK_SMART_PTR_NULL(stream);
1134 0 : std::vector<WrInformation> wrInfoVec;
1135 0 : struct WrAuxInfo aux = {};
1136 0 : HCCL_DEBUG("TX src[%p] len[%llu] dstOffset[%llu]", src, len, dstOffset);
1137 :
1138 0 : if (len > 0) {
1139 0 : CHK_PTR_NULL(src);
1140 0 : CHK_RET(TxPayLoad(dstMemType, dstOffset, src, len, WqeType::WQE_TYPE_DATA, aux, wrInfoVec));
1141 : }
1142 :
1143 0 : CHK_RET(RdmaSendAsync(wrInfoVec, stream, false));
1144 0 : return HCCL_SUCCESS;
1145 0 : }
1146 :
1147 0 : HcclResult TransportDeviceIbverbs::RxData(
1148 : [[maybe_unused]] UserMemType srcMemType, [[maybe_unused]] u64 srcOffset, [[maybe_unused]] void* dst,
1149 : [[maybe_unused]] u64 len, [[maybe_unused]] Stream& stream)
1150 : {
1151 0 : return HCCL_SUCCESS;
1152 : }
1153 :
1154 0 : HcclResult TransportDeviceIbverbs::TxDone(Stream& stream)
1155 : {
1156 : // 发送数据接收确认notify
1157 0 : CHK_RET(TxSendWqe(
1158 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].addr,
1159 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].lkey,
1160 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1161 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream, WqeType::WQE_TYPE_DATA_NOTIFY));
1162 : // 接收数据接收确认notify
1163 0 : CHK_RET(dispatcher_->SignalWait(
1164 : dataAckNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE,
1165 : false, dataAckNotify_->notifyId_));
1166 0 : return HCCL_SUCCESS;
1167 : }
1168 :
1169 0 : HcclResult TransportDeviceIbverbs::RxDone(Stream& stream)
1170 : {
1171 : // 接收数据接收确认notify
1172 0 : CHK_RET(dispatcher_->SignalWait(
1173 : dataNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE,
1174 : false, dataNotify_->notifyId_));
1175 :
1176 : // 发送数据接收确认notify
1177 0 : CHK_RET(TxSendWqe(
1178 : remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].addr,
1179 : remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].lkey,
1180 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1181 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream,
1182 : WqeType::WQE_TYPE_DATA_ACK_NOTIFY));
1183 0 : return HCCL_SUCCESS;
1184 : }
1185 :
1186 0 : HcclResult TransportDeviceIbverbs::PostReady(Stream& stream)
1187 : {
1188 0 : CHK_RET(TxSendWqe(
1189 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].addr,
1190 : remoteMemMsg_[static_cast<u32>(MemType::ACK_NOTIFY_MEM)].lkey,
1191 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1192 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream, WqeType::WQE_TYPE_ACK_NOTIFY));
1193 0 : return HCCL_SUCCESS;
1194 : }
1195 :
1196 0 : HcclResult TransportDeviceIbverbs::WaitReady(Stream& stream)
1197 : {
1198 0 : CHK_RET(dispatcher_->SignalWait(
1199 : ackNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE, false,
1200 : ackNotify_->notifyId_));
1201 0 : return HCCL_SUCCESS;
1202 : }
1203 :
1204 0 : HcclResult TransportDeviceIbverbs::PostFin(Stream& stream)
1205 : {
1206 : // 发送data notify同步信息
1207 0 : void* remoteNotifyaddr = remoteDataNotifyMsg_.addr;
1208 0 : HcclResult ret = TxSendWqe(
1209 0 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].addr,
1210 0 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].lkey,
1211 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1212 0 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream, WqeType::WQE_TYPE_DATA_NOTIFY);
1213 0 : CHK_PRT_RET(
1214 : ret != HCCL_SUCCESS,
1215 : HCCL_ERROR(
1216 : "[TransportDeviceIbverbs][PostFin]errNo[0x%016llx] In ibv tx data signal, send notify "
1217 : "wqe failed. dstMemPtr[%p], srcMemPtr[%p], srcMemSize[%llu]",
1218 : HCCL_ERROR_CODE(ret), remoteNotifyaddr, notifyValueAddr_, notifySize_),
1219 : ret);
1220 : // 每发送一个data notify wqe, count 自增
1221 0 : return HCCL_SUCCESS;
1222 : }
1223 :
1224 0 : HcclResult TransportDeviceIbverbs::WaitFin(Stream& stream)
1225 : {
1226 0 : CHK_RET(dispatcher_->SignalWait(
1227 : dataNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE,
1228 : false, dataNotify_->notifyId_));
1229 0 : return HCCL_SUCCESS;
1230 : }
1231 :
1232 0 : HcclResult TransportDeviceIbverbs::PostFinAck(Stream& stream)
1233 : {
1234 0 : CHK_RET(TxSendWqe(
1235 : remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].addr,
1236 : remoteMemMsg_[static_cast<u32>(MemType::DATA_ACK_NOTIFY_MEM)].lkey,
1237 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1238 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream,
1239 : WqeType::WQE_TYPE_DATA_ACK_NOTIFY));
1240 0 : return HCCL_SUCCESS;
1241 : }
1242 :
1243 0 : HcclResult TransportDeviceIbverbs::WaitFinAck(Stream& stream)
1244 : {
1245 0 : CHK_RET(dispatcher_->SignalWait(
1246 : dataNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE,
1247 : false, dataAckNotify_->notifyId_));
1248 0 : return HCCL_SUCCESS;
1249 : }
1250 :
1251 5 : HcclResult TransportDeviceIbverbs::ResolveRdmaAddrsAndKeys(RdmaAddrKeyResolveParam& param)
1252 : {
1253 5 : param.transLocalAddr = const_cast<void*>(param.localAddr);
1254 5 : param.transRemoteAddr = const_cast<void*>(param.remoteAddr);
1255 5 : if (useMemDetailsLookup_) {
1256 2 : return ResolveRdmaKeysFromMemDetails(param);
1257 : }
1258 3 : return ResolveRdmaKeysFromIoMemRanges(param);
1259 : }
1260 :
1261 2 : HcclResult TransportDeviceIbverbs::ResolveRdmaKeysFromMemDetails(RdmaAddrKeyResolveParam& param)
1262 : {
1263 2 : auto rf = remoteMemDetailsRmaMgr_->Find(MakeMemLookupKey(param.remoteAddr, param.length));
1264 2 : if (!rf.first || rf.second == nullptr) {
1265 1 : HCCL_ERROR(
1266 : "[TransportDeviceIbverbs]Can't find remoteBuffer key by addr and size {%p, %llu}, "
1267 : "registered remote MR count[%zu]",
1268 : param.remoteAddr, param.length, remoteMemDetailsRmaMgr_->size());
1269 1 : return HCCL_E_INTERNAL;
1270 : }
1271 1 : param.dstKey = rf.second->key;
1272 1 : param.transRemoteAddr = LogicalPtrToDevPtr(*rf.second, param.remoteAddr);
1273 :
1274 1 : auto lf = localMemDetailsRmaMgr_->Find(MakeMemLookupKey(param.localAddr, param.length));
1275 1 : CHK_PRT_RET(
1276 : !lf.first || lf.second == nullptr,
1277 : HCCL_ERROR(
1278 : "[TransportDeviceIbverbs]Can't find localBuffer key by addr and size {%p, %llu}, "
1279 : "registered local MR count[%zu]",
1280 : param.localAddr, param.length, localMemDetailsRmaMgr_->size()),
1281 : HCCL_E_INTERNAL);
1282 1 : param.srcKey = lf.second->key;
1283 1 : param.transLocalAddr = LogicalPtrToDevPtr(*lf.second, param.localAddr);
1284 1 : return HCCL_SUCCESS;
1285 2 : }
1286 :
1287 5 : HcclResult TransportDeviceIbverbs::ResolveRdmaKeysFromIoMemRanges(RdmaAddrKeyResolveParam& param)
1288 : {
1289 5 : u64 dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(param.remoteAddr));
1290 : u64 remoteInputAddr
1291 5 : = static_cast<u64>(reinterpret_cast<uintptr_t>(remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].addr));
1292 5 : u64 remoteInputSize = localInputMem_.size;
1293 5 : u32 remoteInputKey = remoteMemMsg_[static_cast<u32>(MemType::USER_INPUT_MEM)].lkey;
1294 : u64 remoteOutputAddr
1295 5 : = static_cast<u64>(reinterpret_cast<uintptr_t>(remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].addr));
1296 5 : u64 remoteOutputSize = localOutputMem_.size;
1297 5 : u32 remoteOutputKey = remoteMemMsg_[static_cast<u32>(MemType::USER_OUTPUT_MEM)].lkey;
1298 5 : if (dstAddr >= remoteInputAddr && dstAddr < remoteInputAddr + remoteInputSize) {
1299 2 : param.dstKey = remoteInputKey;
1300 3 : } else if (dstAddr >= remoteOutputAddr && dstAddr <= remoteOutputAddr + remoteOutputSize) {
1301 1 : param.dstKey = remoteOutputKey;
1302 : } else {
1303 2 : HCCL_ERROR(
1304 : "[TransportDeviceIbverbs][TxAsync]src_ptr=%p is out of range, inputmem src[%p], size[%llu];"
1305 : " outputmem src[%p] size[%llu]",
1306 : param.remoteAddr, remoteInputAddr, remoteInputSize, remoteOutputAddr, remoteOutputSize);
1307 2 : return HCCL_E_INTERNAL;
1308 : }
1309 :
1310 3 : u64 srcAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(param.localAddr));
1311 3 : if (srcAddr >= localInputMem_.addr && srcAddr < localInputMem_.addr + localInputMem_.size) {
1312 2 : param.srcKey = localInputMem_.key;
1313 1 : } else if (srcAddr >= localOutputMem_.addr && srcAddr <= localOutputMem_.addr + localOutputMem_.size) {
1314 1 : param.srcKey = localOutputMem_.key;
1315 : } else {
1316 0 : HCCL_ERROR(
1317 : "[TransportDeviceIbverbs][TxAsync]src_ptr=%p is out of range, inputmem src[%p], size[%llu];"
1318 : " outputmem src[%p] size[%llu]",
1319 : param.localAddr, localInputMem_.addr, localInputMem_.size, localOutputMem_.addr, localOutputMem_.size);
1320 0 : return HCCL_E_INTERNAL;
1321 : }
1322 3 : return HCCL_SUCCESS;
1323 : }
1324 :
1325 0 : HcclResult TransportDeviceIbverbs::WriteCommon(
1326 : const void* remoteAddr, const void* localAddr, u64 length, Stream& stream, WqeType wqeType, struct WrAuxInfo& aux)
1327 : {
1328 0 : if (machinePara_.dctxPtr != nullptr) {
1329 0 : CHK_RET(SetDispatcherCtx(static_cast<DispatcherCtxPtr>(machinePara_.dctxPtr)));
1330 : }
1331 :
1332 0 : std::vector<WrInformation> wrInfoVec;
1333 0 : HCCL_DEBUG("write localAddr[%p] remoteAddr[%p] len[%llu]", localAddr, remoteAddr, length);
1334 :
1335 0 : if (localAddr != nullptr) {
1336 : // 为保证单算子下不同数据量下子图的结构相同，zero byte message 时也需要下发task
1337 0 : u32 txSendDataTimes = (length == 0) ? 1 : (length + RDMA_SEND_MAX_SIZE - 1) / RDMA_SEND_MAX_SIZE;
1338 :
1339 0 : RdmaAddrKeyResolveParam resolve{};
1340 0 : resolve.remoteAddr = remoteAddr;
1341 0 : resolve.localAddr = localAddr;
1342 0 : resolve.length = length;
1343 0 : CHK_RET(ResolveRdmaAddrsAndKeys(resolve));
1344 :
1345 0 : CHK_RET(ConstructPayLoadWqe(
1346 : resolve.transRemoteAddr, resolve.dstKey, resolve.transLocalAddr, resolve.srcKey, length, wqeType, aux,
1347 : wrInfoVec, txSendDataTimes));
1348 : }
1349 0 : u32 maxLength = 0;
1350 0 : for (u32 i = 0; i < wrInfoVec.size(); i++) {
1351 0 : if (wrInfoVec[i].wrData.memList.len > maxLength) {
1352 0 : maxLength = wrInfoVec[i].wrData.memList.len;
1353 : }
1354 : }
1355 :
1356 0 : u32 actualMultiQpNum = GetActualQpNum(maxLength);
1357 :
1358 0 : HCCL_DEBUG(
1359 : "[TransportDeviceIbverbs][TxSendDataAndNotify] UseMultiQp[%d] MultiQpNum[%u] actualMultiQpNum[%u] "
1360 : "maxLength[%u]",
1361 : UseMultiQp(), qpsPerConnection_, actualMultiQpNum, maxLength);
1362 0 : if (UseMultiQp() && actualMultiQpNum != 1 && actualMultiQpNum <= qpsPerConnection_ && maxLength != 0) {
1363 0 : std::vector<std::vector<WrInformation>> multiQpWqeInfoVct(actualMultiQpNum, wrInfoVec);
1364 0 : for (u32 i = 0; i < wrInfoVec.size(); i++) {
1365 0 : WrInformation tmpWqeInfo = wrInfoVec[i];
1366 0 : u32 curLen = tmpWqeInfo.wrData.memList.len;
1367 0 : std::vector<u32> splittedLen = RdmaLengthSplit(curLen, actualMultiQpNum);
1368 0 : uint64_t curSrcAddr = tmpWqeInfo.wrData.memList.addr;
1369 0 : uint64_t curDstAddr = tmpWqeInfo.wrData.dstAddr;
1370 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
1371 0 : multiQpWqeInfoVct[qpIndex][i].wrData.memList.len = splittedLen[qpIndex];
1372 0 : multiQpWqeInfoVct[qpIndex][i].wrData.memList.addr = curSrcAddr;
1373 0 : multiQpWqeInfoVct[qpIndex][i].wrData.dstAddr = curDstAddr;
1374 0 : curSrcAddr += splittedLen[qpIndex];
1375 0 : curDstAddr += splittedLen[qpIndex];
1376 : }
1377 0 : }
1378 :
1379 : // useOneDoorbell 配置成true。最后一个payload去按doorbell
1380 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
1381 0 : CHK_RET(RdmaSendAsync(
1382 : multiQpWqeInfoVct[qpIndex], stream, true, qpIndex)); // 多QP使用同一个stream异步doorbell触发
1383 : }
1384 0 : } else {
1385 0 : CHK_RET(RdmaSendAsync(wrInfoVec, stream, GetUseOneDoorbellValue()));
1386 : }
1387 0 : return HCCL_SUCCESS;
1388 :
1389 : #ifndef CCL_LLT
1390 : CHK_RET(RdmaSendAsync(wrInfoVec, stream, GetUseOneDoorbellValue()));
1391 : #endif
1392 : return HCCL_SUCCESS;
1393 0 : }
1394 :
1395 0 : HcclResult TransportDeviceIbverbs::WriteAsync(
1396 : struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, Stream& stream)
1397 : {
1398 0 : struct WrAuxInfo aux = {};
1399 0 : return WriteCommon(remoteBuf.addr, localBuf.addr, remoteBuf.size, stream, WqeType::WQE_TYPE_DATA, aux);
1400 : }
1401 :
1402 0 : HcclResult TransportDeviceIbverbs::ReadAsync(
1403 : struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, Stream& stream)
1404 : {
1405 0 : HCCL_DEBUG("[TransportDeviceIbverbs][ReadAsync]");
1406 0 : struct WrAuxInfo aux = {};
1407 0 : return WriteCommon(remoteBuf.addr, localBuf.addr, remoteBuf.size, stream, WqeType::WQE_TYPE_READ_DATA, aux);
1408 : }
1409 :
1410 4 : HcclResult TransportDeviceIbverbs::ResolveTransferDesc(
1411 : const HcommBatchTransferDesc& desc, const void*& remoteAddr, const void*& localAddr, u64& length, WqeType& wqeType,
1412 : [[maybe_unused]] struct WrAuxInfo& aux)
1413 : {
1414 4 : if (desc.transType == HCOMM_TRANSFER_TYPE_WRITE) {
1415 3 : CHK_PTR_NULL(desc.transferInfo.write.dst);
1416 1 : CHK_PTR_NULL(desc.transferInfo.write.src);
1417 1 : remoteAddr = desc.transferInfo.write.dst;
1418 1 : localAddr = desc.transferInfo.write.src;
1419 1 : length = desc.transferInfo.write.len;
1420 1 : wqeType = WqeType::WQE_TYPE_DATA;
1421 1 : } else if (desc.transType == HCOMM_TRANSFER_TYPE_READ) {
1422 1 : CHK_PTR_NULL(desc.transferInfo.read.dst);
1423 1 : CHK_PTR_NULL(desc.transferInfo.read.src);
1424 1 : remoteAddr = desc.transferInfo.read.src;
1425 1 : localAddr = desc.transferInfo.read.dst;
1426 1 : length = desc.transferInfo.read.len;
1427 1 : wqeType = WqeType::WQE_TYPE_READ_DATA;
1428 : } else {
1429 0 : HCCL_ERROR("[ResolveTransferDesc] Unsupported transType[%d].", desc.transType);
1430 0 : return HCCL_E_NOT_SUPPORT;
1431 : }
1432 2 : return HCCL_SUCCESS;
1433 : }
1434 :
1435 0 : HcclResult TransportDeviceIbverbs::SubmitWqeBatch(std::vector<WrInformation>& wrInfoVec, Stream& stream)
1436 : {
1437 0 : u32 maxLength = 0;
1438 0 : for (u32 i = 0; i < wrInfoVec.size(); i++) {
1439 0 : if (wrInfoVec[i].wrData.memList.len > maxLength) {
1440 0 : maxLength = wrInfoVec[i].wrData.memList.len;
1441 : }
1442 : }
1443 :
1444 0 : u32 actualMultiQpNum = GetActualQpNum(maxLength);
1445 0 : if (UseMultiQp() && actualMultiQpNum != 1 && actualMultiQpNum <= qpsPerConnection_ && maxLength != 0) {
1446 0 : CHK_RET(TxSendDataAndNotifyWithMultiQP(wrInfoVec, actualMultiQpNum, stream, true));
1447 : } else {
1448 0 : CHK_RET(TxSendDataAndNotifyWithSingleQP(wrInfoVec, stream, true));
1449 : }
1450 0 : return HCCL_SUCCESS;
1451 : }
1452 :
1453 : HcclResult
1454 4 : TransportDeviceIbverbs::BatchTransferImpl(const HcommBatchTransferDesc* transferDescs, uint32_t descNum, Stream& stream)
1455 : {
1456 4 : if (machinePara_.dctxPtr != nullptr) {
1457 0 : CHK_RET(SetDispatcherCtx(static_cast<DispatcherCtxPtr>(machinePara_.dctxPtr)));
1458 : }
1459 :
1460 4 : CHK_PTR_NULL(transferDescs);
1461 4 : std::vector<WrInformation> wrInfoVec;
1462 4 : for (uint32_t i = 0; i < descNum; i++) {
1463 4 : const void* localAddr = nullptr;
1464 4 : const void* remoteAddr = nullptr;
1465 4 : u64 length = 0;
1466 4 : WqeType wqeType = WqeType::WQE_TYPE_DATA;
1467 4 : struct WrAuxInfo aux = {};
1468 6 : CHK_RET(ResolveTransferDesc(transferDescs[i], remoteAddr, localAddr, length, wqeType, aux));
1469 :
1470 2 : HCCL_DEBUG(
1471 : "[BatchTransferImpl] index[%u] localAddr[%p] remoteAddr[%p] len[%llu] wqeType[%d]", i, localAddr,
1472 : remoteAddr, length, static_cast<int>(wqeType));
1473 :
1474 2 : if (localAddr != nullptr) {
1475 2 : u32 txSendDataTimes = (length == 0) ? 1 : (length + RDMA_SEND_MAX_SIZE - 1) / RDMA_SEND_MAX_SIZE;
1476 2 : RdmaAddrKeyResolveParam resolve{};
1477 2 : resolve.remoteAddr = remoteAddr;
1478 2 : resolve.localAddr = localAddr;
1479 2 : resolve.length = length;
1480 2 : CHK_RET(ResolveRdmaAddrsAndKeys(resolve));
1481 0 : CHK_RET(ConstructPayLoadWqe(
1482 : resolve.transRemoteAddr, resolve.dstKey, resolve.transLocalAddr, resolve.srcKey, length, wqeType, aux,
1483 : wrInfoVec, txSendDataTimes));
1484 : }
1485 : }
1486 :
1487 0 : if (!wrInfoVec.empty()) {
1488 0 : return SubmitWqeBatch(wrInfoVec, stream);
1489 : }
1490 0 : return HCCL_SUCCESS;
1491 4 : }
1492 :
1493 4 : HcclResult TransportDeviceIbverbs::BatchTransferAsync(
1494 : const HcommBatchTransferDesc* transferDescs, uint32_t descNum, Stream& stream)
1495 : {
1496 4 : return BatchTransferImpl(transferDescs, descNum, stream);
1497 : }
1498 :
1499 0 : HcclResult TransportDeviceIbverbs::WriteReduceAsync(
1500 : struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, const HcclDataType datatype,
1501 : HcclReduceOp redOp, Stream& stream)
1502 : {
1503 0 : struct WrAuxInfo aux = {};
1504 0 : aux.dataType = RDMA_REDUCE_DATA_TYPE_TABLE[datatype];
1505 0 : aux.reduceType = RDMA_REDUCE_OP_TYPE_TABLE[redOp];
1506 0 : if (aux.dataType == static_cast<uint8_t>(RdmaReduceDataType::RDMA_REDUCE_DATA_INVALID)
1507 0 : || aux.reduceType == static_cast<uint8_t>(RdmaReduceOpType::RDMA_REDUCE_OP_INVALID)) {
1508 0 : HCCL_ERROR(
1509 : "unsupported data type [%s] or Reduce type [%s]", GetDataTypeEnumStr(datatype).c_str(),
1510 : GetReduceOpEnumStr(redOp).c_str());
1511 0 : return HCCL_E_INTERNAL;
1512 : }
1513 :
1514 0 : return WriteCommon(remoteBuf.addr, localBuf.addr, remoteBuf.size, stream, WqeType::WQE_TYPE_DATA_WITH_REDUCE, aux);
1515 : }
1516 :
1517 0 : HcclResult TransportDeviceIbverbs::Post(u32 notifyIdx, Stream& stream)
1518 : {
1519 : // 校验notifyIdx有效性
1520 0 : bool bRet = (notifyIdx >= notifyNum_);
1521 0 : CHK_PRT_RET(
1522 : bRet,
1523 : HCCL_ERROR(
1524 : "[TransportDeviceIbverbs][Post]notifyNum[%u], notifyIdx[%u] out of range[0, %u]", notifyNum_, notifyIdx,
1525 : notifyNum_ - 1),
1526 : HCCL_E_INTERNAL);
1527 :
1528 : // 每个QP发送一个指定idx的notify
1529 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
1530 0 : CHK_RET(TxSendWqe(
1531 : userMultiQpRemoteNotifyMsg_[i][notifyIdx].addr, userMultiQpRemoteNotifyMsg_[i][notifyIdx].lkey,
1532 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
1533 : memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey, notifySize_, stream,
1534 : WqeType::WQE_TYPE_DATA_WITH_NOTIFY));
1535 : }
1536 0 : return HCCL_SUCCESS;
1537 : }
1538 :
1539 0 : HcclResult TransportDeviceIbverbs::Wait(u32 notifyIdx, Stream& stream, const u32 timeOut)
1540 : {
1541 : // 校验notifyIdx有效性
1542 0 : bool bRet = (notifyIdx >= notifyNum_);
1543 0 : CHK_PRT_RET(
1544 : bRet,
1545 : HCCL_ERROR(
1546 : "[TransportDeviceIbverbs][Wait]notifyNum[%u], notifyIdx[%u] out of range[0, %u]", notifyNum_, notifyIdx,
1547 : notifyNum_ - 1),
1548 : HCCL_E_INTERNAL);
1549 :
1550 : // 单QP接收一个指定idx的notify
1551 : // 每个qp接收一个指定idx的notify
1552 0 : for (u32 i = 0; i < qpsPerConnection_; i++) {
1553 0 : CHK_RET(dispatcher_->SignalWait(
1554 : userMultiQpLocalNotify_[i][notifyIdx]->ptr(), stream, machinePara_.localUserrank,
1555 : machinePara_.remoteWorldRank, INVALID_VALUE_STAGE, false, userMultiQpLocalNotify_[i][notifyIdx]->notifyId_,
1556 : timeOut));
1557 : }
1558 0 : return HCCL_SUCCESS;
1559 : }
1560 :
1561 0 : bool TransportDeviceIbverbs::UseMultiQp() { return qpsPerConnection_ != 1; }
1562 :
1563 0 : u32 TransportDeviceIbverbs::GetActualQpNum(u32 maxLength)
1564 : {
1565 0 : u32 actualMultiQpNum = 1;
1566 0 : const u32 KByteToByte = 1024; // 1024 多QP阈值单位是KB
1567 0 : if (maxLength / qpsPerConnection_ >= multiQpThreshold_ * KByteToByte) {
1568 0 : actualMultiQpNum = qpsPerConnection_;
1569 : } else {
1570 0 : u32 quotient = maxLength / (multiQpThreshold_ * KByteToByte);
1571 0 : u32 remainder = maxLength % (multiQpThreshold_ * KByteToByte);
1572 0 : actualMultiQpNum = quotient + (remainder != 0 ? 1 : 0);
1573 : }
1574 :
1575 0 : return actualMultiQpNum;
1576 : }
1577 :
1578 0 : HcclResult TransportDeviceIbverbs::TxSendDataAndNotifyWithMultiQP(
1579 : std::vector<WrInformation>& wqeInfoVec, u32 actualMultiQpNum, Stream& stream, [[maybe_unused]] bool useOneDoorbell)
1580 : {
1581 : // vector<WrInformation> 是一个vector的原因是 单个wqe只能发2GB数据，如果超过2GB，就拆分到多个WqeInfo中了
1582 : // 多QP下，对每个WqeInfo都进行多QP切分，然后在收发每一个QP的数据
1583 0 : std::vector<std::vector<WrInformation>> multiQpWqeInfoVct(actualMultiQpNum, wqeInfoVec);
1584 0 : for (u32 i = 0; i < wqeInfoVec.size(); i++) {
1585 0 : WrInformation tmpWqeInfo = wqeInfoVec[i];
1586 0 : u32 curLen = tmpWqeInfo.wrData.memList.len;
1587 0 : std::vector<u32> splittedLen = RdmaLengthSplit(curLen, actualMultiQpNum);
1588 0 : uint64_t curSrcAddr = tmpWqeInfo.wrData.memList.addr;
1589 0 : uint64_t curDstAddr = tmpWqeInfo.wrData.dstAddr;
1590 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
1591 0 : multiQpWqeInfoVct[qpIndex][i].wrData.memList.len = splittedLen[qpIndex];
1592 0 : multiQpWqeInfoVct[qpIndex][i].wrData.memList.addr = curSrcAddr;
1593 0 : multiQpWqeInfoVct[qpIndex][i].wrData.dstAddr = curDstAddr;
1594 0 : curSrcAddr += splittedLen[qpIndex];
1595 0 : curDstAddr += splittedLen[qpIndex];
1596 : }
1597 0 : }
1598 : // 给每个QP最后增加一个属于该QP的DataNotify
1599 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
1600 0 : struct WrAuxInfo aux = {};
1601 0 : void* remoteNotifyaddr = multiQpDataNotifyRemoteMemMsg_[qpIndex].addr;
1602 0 : CHK_RET(AddWrList(
1603 : remoteNotifyaddr, notifyValueAddr_, notifySize_, memMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey,
1604 : remoteMemMsg_[static_cast<u32>(MemType::DATA_NOTIFY_MEM)].lkey, WqeType::WQE_TYPE_DATA_NOTIFY, aux,
1605 : multiQpWqeInfoVct[qpIndex]));
1606 : }
1607 : // useOneDoorbell 配置成true。最后一个payload去按doorbell
1608 0 : for (u32 qpIndex = 0; qpIndex < actualMultiQpNum; qpIndex++) {
1609 0 : CHK_RET(
1610 : RdmaSendAsync(multiQpWqeInfoVct[qpIndex], stream, true, qpIndex)); // 多QP使用同一个stream异步doorbell触发
1611 : }
1612 0 : return HCCL_SUCCESS;
1613 0 : }
1614 0 : HcclResult TransportDeviceIbverbs::GetTransportId(u32& id)
1615 : {
1616 0 : struct ibv_qp* qp = reinterpret_cast<struct ibv_qp*>(combineAiQpInfo_.aiQpInfo.aiQpAddr);
1617 0 : if (nullptr != qp) {
1618 0 : id = qp->qp_num;
1619 : }
1620 0 : return HCCL_SUCCESS;
1621 : }
1622 :
1623 0 : HcclResult TransportDeviceIbverbs::HnsPostSend(
1624 : const TransportDeviceNormalData& ibvData, struct MemDetails* localMems, struct MemDetails* remoteMems, u32 memNum,
1625 : HcclWrOpCode opCode, u64& dbInfo, bool fence)
1626 : {
1627 0 : CHK_PTR_NULL(localMems);
1628 0 : CHK_PTR_NULL(remoteMems);
1629 :
1630 0 : const uint32_t SEND_WR_LEN = 8;
1631 0 : uint32_t last = memNum - 1;
1632 0 : CHK_PRT_RET(
1633 : memNum > SEND_WR_LEN,
1634 : HCCL_ERROR("[TransportDeviceIbverbs][HnsPostSend] buffer size is:%u over SEND_WR_LEN: %u", memNum, SEND_WR_LEN),
1635 : HCCL_E_PARA);
1636 0 : struct ibv_send_wr sendWr[SEND_WR_LEN] = {};
1637 0 : struct ibv_sge sge[SEND_WR_LEN] = {};
1638 :
1639 0 : for (uint32_t index = 0; index < memNum; index++) {
1640 : // 设置WR的SGE
1641 0 : sge[index].addr = reinterpret_cast<u64>(localMems[index].addr);
1642 0 : sge[index].length = remoteMems[index].size;
1643 0 : sge[index].lkey = localMems[index].key;
1644 :
1645 : // 设置WR属性
1646 0 : sendWr[index].wr_id = wrIdOffset_.fetch_add(1, std::memory_order_relaxed);
1647 0 : sendWr[index].num_sge = 1; // 只有一个SGE
1648 0 : sendWr[index].sg_list = &sge[index];
1649 0 : sendWr[index].wr.rdma.remote_addr = reinterpret_cast<u64>(remoteMems[index].addr);
1650 0 : sendWr[index].wr.rdma.rkey = remoteMems[index].key;
1651 0 : sendWr[index].next = (index == last) ? nullptr : &sendWr[index + 1]; // 第一个WR指向第二个WR
1652 0 : sendWr[index].send_flags = (index == last) ?
1653 : (fence ? (IBV_SEND_SIGNALED | IBV_SEND_FENCE) : IBV_SEND_SIGNALED) :
1654 : 0; // 最后一个WR才需要回复CQE
1655 0 : sendWr[index].opcode = static_cast<enum ibv_wr_opcode>(opCode);
1656 0 : HCCL_DEBUG(
1657 : "[TransportDeviceIbverbs][HnsPostSend] Direct ibv_post_send[%llu], opcode=[0x%x], "
1658 : "remote_addr=[0x%llx], size=[%u], fence[%u]",
1659 : wrIdOffset_.load(), sendWr[index].opcode, sendWr[index].wr.rdma.remote_addr, sendWr[index].sg_list->length,
1660 : fence);
1661 : }
1662 :
1663 0 : struct ibv_send_wr* badWr = nullptr;
1664 0 : struct WrExpRsp exp_rsp = {};
1665 0 : struct ibv_qp* qp = reinterpret_cast<struct ibv_qp*>(ibvData.qpInfo.qpPtr);
1666 0 : CHK_PTR_NULL(qp);
1667 0 : HCCL_DEBUG(
1668 : "[TransportDeviceIbverbs][HnsPostSend] qp=%p, handle=%u, qp_num=%u, qp_type=%d, qp_stat=%d", qp, qp->handle,
1669 : qp->qp_num, qp->qp_type, qp->state);
1670 0 : HcclResult ret = HrtHnsIbvExpPostSend(qp, &sendWr[0], &badWr, &exp_rsp);
1671 0 : HCOMM_DSB();
1672 0 : CHK_PRT_RET(
1673 : ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN,
1674 : HCCL_ERROR(
1675 : "[TransportDeviceIbverbs][HnsPostSend] failed, qp=%p, handle=%u, qp_num=%u, qp_type=%d, qp_stat=%d", qp,
1676 : qp->handle, qp->qp_num, qp->qp_type, qp->state),
1677 : ret);
1678 0 : if (ret == HCCL_SUCCESS) {
1679 0 : dbInfo = exp_rsp.db_info;
1680 : }
1681 0 : return ret;
1682 : }
1683 :
1684 1 : HcclResult TransportDeviceIbverbs::Drain(Stream& stream)
1685 : {
1686 1 : CHK_PTR_NULL(dataNotify_);
1687 1 : CHK_PTR_NULL(remoteMemMsg_[MemType::NOTIFY_SRC_MEM].addr);
1688 1 : CHK_PTR_NULL(memMsg_[MemType::DATA_NOTIFY_MEM].addr);
1689 :
1690 1 : struct SgList list = {};
1691 1 : struct SendWr wr = {};
1692 1 : list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(memMsg_[MemType::DATA_NOTIFY_MEM].addr));
1693 1 : list.len = static_cast<u32>(memMsg_[MemType::DATA_NOTIFY_MEM].len);
1694 1 : list.lkey = memMsg_[MemType::DATA_NOTIFY_MEM].lkey;
1695 :
1696 1 : wr.bufList = &list;
1697 1 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
1698 1 : wr.dstAddr = reinterpret_cast<u64>(remoteMemMsg_[MemType::NOTIFY_SRC_MEM].addr);
1699 1 : wr.rkey = remoteMemMsg_[MemType::NOTIFY_SRC_MEM].lkey;
1700 1 : wr.op = RaWrOpcode::RA_WR_RDMA_READ;
1701 1 : wr.sendFlag = RA_SEND_SIGNALED | RA_SEND_FENCE; // fence
1702 :
1703 1 : CHK_RET(RdmaSendAsync(wr, stream, WqeType::WQE_TYPE_DATA_NOTIFY, wr.dstAddr, INVALID_UINT));
1704 1 : CHK_RET(dispatcher_->SignalWait(
1705 : dataNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank, INVALID_VALUE_STAGE,
1706 : false, dataNotify_->notifyId_));
1707 1 : return HCCL_SUCCESS;
1708 : }
1709 : } // namespace hccl
|
