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