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 "transport_device_roce_mem.h"
12 : #include "log.h"
13 : #include "dispatcher_pub.h"
14 : #include "adapter_verbs.h"
15 :
16 : namespace hccl {
17 : std::atomic<u64> TransportDeviceRoceMem::wrIdOffset_;
18 :
19 0 : TransportDeviceRoceMem::TransportDeviceRoceMem(
20 : const std::unique_ptr<NotifyPool>& notifyPool, const HcclNetDevCtx& netDevCtx, const HcclDispatcher& dispatcher,
21 0 : AttrInfo& attrInfo, bool aicpuUnfoldMode, const HcclQpInfoV2& qpInfo)
22 : : TransportMem(notifyPool, netDevCtx, dispatcher, attrInfo, aicpuUnfoldMode),
23 0 : timeout_{std::chrono::microseconds((attrInfo.timeout == INVALID_UINT) ? 0 : attrInfo.timeout)},
24 0 : qpInfo_{qpInfo}
25 0 : {}
26 :
27 0 : TransportDeviceRoceMem::~TransportDeviceRoceMem() {}
28 :
29 0 : HcclResult TransportDeviceRoceMem::ExchangeMemDesc(
30 : [[maybe_unused]] const RmaMemDescs& localMemDescs, [[maybe_unused]] RmaMemDescs& remoteMemDescs,
31 : [[maybe_unused]] u32& actualNumOfRemote)
32 : {
33 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support ExchangeMemDesc");
34 0 : return HCCL_E_NOT_SUPPORT;
35 : }
36 :
37 0 : HcclResult TransportDeviceRoceMem::EnableMemAccess(
38 : [[maybe_unused]] const RmaMemDesc& remoteMemDesc, [[maybe_unused]] RmaMem& remoteMem)
39 : {
40 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support EnableMemAccess");
41 0 : return HCCL_E_NOT_SUPPORT;
42 : }
43 :
44 0 : HcclResult TransportDeviceRoceMem::DisableMemAccess([[maybe_unused]] const RmaMemDesc& remoteMemDesc)
45 : {
46 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support DisableMemAccess");
47 0 : return HCCL_E_NOT_SUPPORT;
48 : }
49 :
50 0 : HcclResult TransportDeviceRoceMem::SetSocket([[maybe_unused]] const std::shared_ptr<HcclSocket>& socket)
51 : {
52 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support SetSocket");
53 0 : return HCCL_E_NOT_SUPPORT;
54 : }
55 :
56 0 : HcclResult TransportDeviceRoceMem::Connect([[maybe_unused]] s32 timeoutSec)
57 : {
58 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support Connect");
59 0 : return HCCL_E_NOT_SUPPORT;
60 : }
61 :
62 0 : HcclResult TransportDeviceRoceMem::Write(
63 : [[maybe_unused]] const HcclBuf& remoteMem, [[maybe_unused]] const HcclBuf& localMem,
64 : [[maybe_unused]] const rtStream_t& stream)
65 : {
66 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support HcclBuf Write");
67 0 : return HCCL_E_NOT_SUPPORT;
68 : }
69 :
70 0 : HcclResult TransportDeviceRoceMem::Read(
71 : [[maybe_unused]] const HcclBuf& localMem, [[maybe_unused]] const HcclBuf& remoteMem,
72 : [[maybe_unused]] const rtStream_t& stream)
73 : {
74 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support HcclBuf Read");
75 0 : return HCCL_E_NOT_SUPPORT;
76 : }
77 :
78 0 : HcclResult TransportDeviceRoceMem::Write(
79 : [[maybe_unused]] const RmaOpMem& remoteMem, [[maybe_unused]] const RmaOpMem& localMem,
80 : [[maybe_unused]] const rtStream_t& stream)
81 : {
82 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support RmaOpMem Write");
83 0 : return HCCL_E_NOT_SUPPORT;
84 : }
85 :
86 0 : HcclResult TransportDeviceRoceMem::Read(
87 : [[maybe_unused]] const RmaOpMem& localMem, [[maybe_unused]] const RmaOpMem& remoteMem,
88 : [[maybe_unused]] const rtStream_t& stream)
89 : {
90 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support RmaOpMem Read");
91 0 : return HCCL_E_NOT_SUPPORT;
92 : }
93 :
94 0 : HcclResult TransportDeviceRoceMem::AddOpFence([[maybe_unused]] const rtStream_t& stream)
95 : {
96 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support HOST AddOpFence");
97 0 : return HCCL_E_NOT_SUPPORT;
98 : }
99 :
100 0 : HcclResult TransportDeviceRoceMem::GetTransInfo(
101 : [[maybe_unused]] HcclQpInfoV2& qpInfo, [[maybe_unused]] u32* lkey, [[maybe_unused]] u32* rkey,
102 : [[maybe_unused]] HcclBuf* localMem, [[maybe_unused]] HcclBuf* remoteMem, [[maybe_unused]] u32 num)
103 : {
104 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support GetTransInfo");
105 0 : return HCCL_E_NOT_SUPPORT;
106 : }
107 :
108 0 : HcclResult TransportDeviceRoceMem::WaitOpFence([[maybe_unused]] const rtStream_t& stream)
109 : {
110 0 : HCCL_ERROR("TransportDeviceRoceMem doesn't support WaitOpFence");
111 0 : return HCCL_E_NOT_SUPPORT;
112 : }
113 :
114 0 : HcclResult TransportDeviceRoceMem::TransportDeviceRoceMem::BatchWrite(
115 : const std::vector<MemDetails>& remoteMems, const std::vector<MemDetails>& localMems, Stream& stream)
116 : {
117 0 : return BatchOp(stream, localMems, remoteMems, false, false);
118 : }
119 :
120 0 : HcclResult TransportDeviceRoceMem::BatchRead(
121 : const std::vector<MemDetails>& localMems, const std::vector<MemDetails>& remoteMems, Stream& stream)
122 : {
123 0 : return BatchOp(stream, localMems, remoteMems, true, false);
124 : }
125 :
126 : HcclResult
127 0 : TransportDeviceRoceMem::AddOpFence(const MemDetails& localFenceMem, const MemDetails& remoteFenceMem, Stream& stream)
128 : {
129 0 : std::vector<MemDetails> localMems(1, localFenceMem);
130 0 : std::vector<MemDetails> remoteMems(1, remoteFenceMem);
131 0 : return BatchOp(stream, localMems, remoteMems, true, true);
132 0 : }
133 :
134 0 : HcclResult TransportDeviceRoceMem::DoorBellSend(Stream& stream, u64 dbInfo, u32 wrDataLen, bool fence)
135 : {
136 0 : HCCL_DEBUG("[DoorBellSend] dbIndex[%#x] dbInfo[%#llx] remoteRankId[%u]", qpInfo_.dbIndex, dbInfo, remoteRankId_);
137 0 : RdmaTaskInfo rdmaInfo;
138 0 : WrInformation wrInfo;
139 0 : wrInfo.notifyId = 0;
140 0 : wrInfo.wrData.memList.len = wrDataLen;
141 0 : rdmaInfo.wrInfos.emplace_back(wrInfo);
142 0 : rdmaInfo.rdmaType = fence ? RdmaType::RDMA_SEND_NOTIFY : RdmaType::RDMA_SEND_PAYLOAD;
143 0 : rdmaInfo.remoteRank = remoteRankId_;
144 0 : HcclResult ret = HCCL_SUCCESS;
145 0 : DispatcherPub* dispatcher = static_cast<DispatcherPub*>(dispatcher_);
146 0 : ret = dispatcher->RdmaSend(qpInfo_.dbIndex, dbInfo, stream, rdmaInfo);
147 0 : CHK_PRT_RET(
148 : ret != HCCL_SUCCESS,
149 : HCCL_ERROR(
150 : "[DoorBellSend] RdmaSend failed, ret[%u]. dbIndex[%#x] dbInfo[%#llx] remoteRankId[%u]", ret,
151 : qpInfo_.dbIndex, dbInfo, rdmaInfo.remoteRank),
152 : ret);
153 0 : std::vector<Stream> subStreams;
154 0 : ret = dispatcher->LaunchTasksEx(stream, subStreams);
155 0 : CHK_PRT_RET(
156 : ret != HCCL_SUCCESS,
157 : HCCL_ERROR(
158 : "[DoorBellSend] LaunchTask failed, ret[%u]. dbIndex[%#x] dbInfo[%#llx] remoteRankId[%u]", ret,
159 : qpInfo_.dbIndex, dbInfo, rdmaInfo.remoteRank),
160 : ret);
161 0 : return HCCL_SUCCESS;
162 0 : }
163 :
164 0 : HcclResult TransportDeviceRoceMem::FillMemDetails(
165 : std::vector<MemDetails>& localMemList, std::vector<MemDetails>& remoteMemList, MemDetails& localMem,
166 : MemDetails& remoteMem)
167 : {
168 0 : CHK_PRT_RET(
169 : localMem.size != remoteMem.size,
170 : HCCL_ERROR(
171 : "[TransportDeviceRoceMem][FillMemDetails] "
172 : "local buffer size[%llu] is not equal to remote buffer size[%llu]",
173 : localMem.size, remoteMem.size),
174 : HCCL_E_PARA);
175 0 : u64 remainingBytes = localMem.size;
176 0 : while (remainingBytes > 0) {
177 0 : const u64 chunkBytes = (remainingBytes > MAX_RDMA_WQE_SIZE) ? MAX_RDMA_WQE_SIZE : remainingBytes;
178 0 : localMem.size = chunkBytes;
179 0 : remoteMem.size = chunkBytes;
180 0 : localMemList.emplace_back(localMem);
181 0 : remoteMemList.emplace_back(remoteMem);
182 0 : localMem.addr += chunkBytes;
183 0 : remoteMem.addr += chunkBytes;
184 0 : remainingBytes -= chunkBytes;
185 : }
186 0 : return HCCL_SUCCESS;
187 : }
188 :
189 0 : HcclResult TransportDeviceRoceMem::BatchOp(
190 : Stream& stream, const std::vector<MemDetails>& localMems, const std::vector<MemDetails>& remoteMems, bool isRead,
191 : bool fence)
192 : {
193 0 : constexpr u32 MAX_RDMA_WQE_NUM = 64; // related to qp depth
194 0 : CHK_PRT_RET(
195 : localMems.size() != remoteMems.size(),
196 : HCCL_ERROR(
197 : "[TransportDeviceRoceMem][BatchOp] "
198 : "local buffer num[%llu] is not equal to remote buffer num[%llu]",
199 : localMems.size(), remoteMems.size()),
200 : HCCL_E_PARA);
201 0 : u64 dbInfo = 0;
202 0 : u32 wqeCount = 0;
203 0 : u64 wrDataLen = 0;
204 0 : const u32 memNum = localMems.size();
205 0 : for (u32 index = 0; index < memNum; index++) {
206 0 : MemDetails localMem = localMems[index];
207 0 : MemDetails remoteMem = remoteMems[index];
208 0 : wrDataLen += localMem.size;
209 0 : std::vector<MemDetails> localMemList;
210 0 : std::vector<MemDetails> remoteMemList;
211 0 : CHK_RET(FillMemDetails(localMemList, remoteMemList, localMem, remoteMem));
212 0 : CHK_RET(BatchPostSend(stream, dbInfo, localMemList, remoteMemList, isRead, fence, wqeCount, wrDataLen));
213 0 : if (wqeCount >= MAX_RDMA_WQE_NUM) {
214 0 : CHK_RET(DoorBellSend(stream, dbInfo, wrDataLen, fence));
215 0 : wqeCount = 0;
216 0 : wrDataLen = 0;
217 : }
218 0 : }
219 0 : if (wqeCount != 0) {
220 0 : CHK_RET(DoorBellSend(stream, dbInfo, wrDataLen, fence));
221 : }
222 0 : return HCCL_SUCCESS;
223 : }
224 :
225 0 : HcclResult TransportDeviceRoceMem::BatchPostSend(
226 : Stream& stream, u64& dbInfo, std::vector<MemDetails>& localMemList, std::vector<MemDetails>& remoteMemList,
227 : bool isRead, bool fence, u32& wqeCount, u64& wrDataLen)
228 : {
229 0 : const u32 wrTotalCount = localMemList.size();
230 0 : u32 sendWrCount = 0;
231 0 : while (sendWrCount < wrTotalCount) {
232 0 : const u32 wrCount = std::min(wrTotalCount - sendWrCount, SEND_WR_LEN);
233 0 : CHK_RET(PostSend(
234 : stream, dbInfo, &(localMemList[sendWrCount]), &(remoteMemList[sendWrCount]), wrCount, isRead, fence,
235 : wqeCount, wrDataLen));
236 0 : sendWrCount += wrCount;
237 0 : wqeCount += wrCount;
238 : }
239 0 : return HCCL_SUCCESS;
240 : }
241 :
242 0 : HcclResult TransportDeviceRoceMem::PostSend(
243 : Stream& stream, u64& dbInfo, struct MemDetails* localMems, struct MemDetails* remoteMems, u32 memNum, bool isRead,
244 : bool fence, u32& wqeCount, u64& wrDataLen)
245 : {
246 0 : constexpr u32 RETRY_DELAY_THRESH = 100;
247 0 : u32 retryCount = 0;
248 0 : auto startTime = std::chrono::steady_clock::now();
249 0 : HcclResult ret = HCCL_E_NETWORK;
250 0 : while (ret != HCCL_SUCCESS) {
251 0 : RdmaOp opCode = isRead ? RdmaOp::OP_READ : RdmaOp::OP_WRITE;
252 0 : ret = RdmaPostSend(dbInfo, localMems, remoteMems, memNum, opCode, fence);
253 0 : if (ret == HCCL_E_AGAIN) {
254 0 : if ((retryCount == 0) && (wqeCount != 0)) {
255 0 : HCCL_WARNING(
256 : "[PostSend] retry with DoorBellSend, isRead[%u] remoteRankId[%u] wqeCount[%u]", isRead,
257 : remoteRankId_, wqeCount);
258 0 : CHK_RET(DoorBellSend(stream, dbInfo, wrDataLen, fence));
259 0 : wqeCount = 0;
260 0 : wrDataLen = 0;
261 0 : } else {
262 0 : CHK_PRT_RET(
263 : timeout_ == std::chrono::microseconds(0),
264 : HCCL_ERROR("[PostSend] failed without retry, isRead[%u] remoteRankId[%u]", isRead, remoteRankId_),
265 : ret);
266 0 : auto elapsedTime = std::chrono::duration_cast<std::chrono::microseconds>(
267 0 : std::chrono::steady_clock::now() - startTime);
268 0 : CHK_PRT_RET(
269 : elapsedTime >= timeout_,
270 : HCCL_ERROR(
271 : "[PostSend] failed after timeout, elapsedTime[%lld us] isRead[%u] remoteRankId[%u]",
272 : elapsedTime.count(), isRead, remoteRankId_),
273 : ret);
274 0 : if (retryCount % RETRY_DELAY_THRESH == 0) {
275 0 : HCCL_WARNING(
276 : "[PostSend] retryCount[%u] after failed, elapsedTime[%lld us] isRead[%u] "
277 : "remoteRankId[%u]",
278 : retryCount, elapsedTime.count(), isRead, remoteRankId_);
279 : }
280 0 : SaluSleep(
281 0 : ONE_MILLISECOND_OF_USLEEP * std::min(CeilDiv(retryCount, RETRY_DELAY_THRESH), RETRY_DELAY_THRESH));
282 : }
283 0 : ++retryCount;
284 0 : continue; // to retry
285 0 : }
286 0 : CHK_PRT_RET(
287 : ret != HCCL_SUCCESS,
288 : HCCL_ERROR("[PostSend] HnsPostSend failed[%u], isRead[%u] remoteRankId[%u]", ret, isRead, remoteRankId_),
289 : ret);
290 0 : if (retryCount != 0) {
291 0 : HCCL_INFO("[PostSend] retry success, isRead[%u] remoteRankId[%u]", isRead, remoteRankId_);
292 : }
293 : }
294 0 : return HCCL_SUCCESS;
295 : }
296 :
297 0 : HcclResult TransportDeviceRoceMem::RdmaPostSend(
298 : u64& dbInfo, MemDetails* localMems, MemDetails* remoteMems, u32 memNum, RdmaOp opCode, bool fence)
299 : {
300 0 : CHK_PTR_NULL(localMems);
301 0 : CHK_PTR_NULL(remoteMems);
302 :
303 0 : CHK_PRT_RET(
304 : memNum > SEND_WR_LEN,
305 : HCCL_ERROR(
306 : "[TransportDeviceRoceMem][RdmaPostSend] buffer size is:%u over SEND_WR_LEN: %u", memNum, SEND_WR_LEN),
307 : HCCL_E_PARA);
308 0 : const u32 last = memNum - 1;
309 0 : struct ibv_send_wr sendWr[SEND_WR_LEN] = {};
310 0 : struct ibv_sge sge[SEND_WR_LEN] = {};
311 0 : for (u32 index = 0; index < memNum; index++) {
312 : // 设置WR的SGE
313 0 : sge[index].addr = localMems[index].addr;
314 0 : sge[index].length = remoteMems[index].size;
315 0 : sge[index].lkey = localMems[index].key;
316 :
317 : // 设置WR属性
318 0 : sendWr[index].wr_id = wrIdOffset_.fetch_add(1, std::memory_order_relaxed);
319 0 : sendWr[index].num_sge = 1; // 只有一个SGE
320 0 : sendWr[index].sg_list = &sge[index];
321 0 : sendWr[index].wr.rdma.remote_addr = remoteMems[index].addr;
322 0 : sendWr[index].wr.rdma.rkey = remoteMems[index].key;
323 0 : sendWr[index].next = (index == last) ? nullptr : &sendWr[index + 1]; // 第一个WR指向第二个WR
324 0 : sendWr[index].send_flags = (index == last) ?
325 : (fence ? (IBV_SEND_SIGNALED | IBV_SEND_FENCE) : IBV_SEND_SIGNALED) :
326 : 0; // 最后一个WR才需要回复CQE
327 0 : sendWr[index].opcode = static_cast<enum ibv_wr_opcode>(opCode);
328 0 : HCCL_DEBUG(
329 : "[TransportDeviceRoceMem][RdmaPostSend] Direct ibv_post_send[%llu], opcode=[0x%x], "
330 : "remote_addr=[0x%llx], size=[%u], fence[%u]",
331 : wrIdOffset_.load(), sendWr[index].opcode, sendWr[index].wr.rdma.remote_addr, sendWr[index].sg_list->length,
332 : fence);
333 : }
334 :
335 0 : struct ibv_send_wr* badWr = nullptr;
336 0 : struct WrExpRsp exp_rsp = {};
337 0 : struct ibv_qp* qp = reinterpret_cast<struct ibv_qp*>(qpInfo_.qpPtr);
338 0 : CHK_PTR_NULL(qp);
339 0 : HCCL_DEBUG(
340 : "[TransportDeviceRoceMem][RdmaPostSend] qp=%p, handle=%u, qp_num=%u, qp_type=%d, qp_stat=%d", qp, qp->handle,
341 : qp->qp_num, qp->qp_type, qp->state);
342 0 : HcclResult ret = HrtHnsIbvExpPostSend(qp, &sendWr[0], &badWr, &exp_rsp);
343 0 : CHK_PRT_RET(
344 : ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN,
345 : HCCL_ERROR(
346 : "[TransportDeviceRoceMem][RdmaPostSend] failed, qp=%p, handle=%u, qp_num=%u, qp_type=%d, qp_stat=%d", qp,
347 : qp->handle, qp->qp_num, qp->qp_type, qp->state),
348 : ret);
349 0 : if (ret == HCCL_SUCCESS) {
350 0 : dbInfo = exp_rsp.db_info;
351 : }
352 0 : return ret;
353 0 : }
354 : } // namespace hccl
|
