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 "send_recv_executor.h"
12 : #include "device_capacity.h"
13 : #include "dispatcher_pub.h"
14 : #include "adapter_rts.h"
15 : #include "externalinput_pub.h"
16 :
17 : namespace hccl {
18 : constexpr u32 HCCL_POLL_CQ_DEPTH = 32;
19 : constexpr s32 HCCL_POLL_CQ_ONETIME = 1;
20 : constexpr u32 HCCL_POLL_CQ_INTERVAL = 100;
21 : constexpr u64 MAX_RDMA_WQE_SIZE = 2ULL * 1024 * 1024 * 1024; // RDMA最大WQE限制, 2G限制是RDMA导致
22 :
23 172 : struct MrInfoT AscendMrInfo2MrInfo(AscendMrInfo* ascendMrInfo)
24 : {
25 172 : struct MrInfoT innerMrInfo = {};
26 172 : innerMrInfo.addr = reinterpret_cast<void*>(ascendMrInfo->addr);
27 172 : innerMrInfo.size = ascendMrInfo->size;
28 172 : innerMrInfo.lkey = ascendMrInfo->key;
29 172 : return innerMrInfo;
30 : }
31 :
32 0 : SendRecvExecutor::SendRecvExecutor(
33 : HcclRtStream stream, QpHandle qpHandle, const struct MrInfoT& localWindowMem, const struct MrInfoT& remoteWindowMem,
34 : const struct MrInfoT& localSyncMemPrepare, const struct MrInfoT& localSyncMemDone,
35 : const struct MrInfoT& localSyncMemAck, const struct MrInfoT& remoteSyncMemPrepare,
36 0 : const struct MrInfoT& remoteSyncMemDone, const struct MrInfoT& remoteSyncMemAck, u32 immData, const u64 chunkNum)
37 0 : : stream_(stream),
38 0 : qpHandle_(qpHandle),
39 0 : localWindowMem_(localWindowMem),
40 0 : remoteWindowMem_(remoteWindowMem),
41 0 : localSyncMemPrepare_(localSyncMemPrepare),
42 0 : localSyncMemDone_(localSyncMemDone),
43 0 : localSyncMemAck_(localSyncMemAck),
44 0 : remoteSyncMemPrepare_(remoteSyncMemPrepare),
45 0 : remoteSyncMemDone_(remoteSyncMemDone),
46 0 : remoteSyncMemAck_(remoteSyncMemAck),
47 0 : immData_(immData),
48 0 : chunkSize_(chunkNum),
49 0 : notifyWaitMode_(SyncMode::DEFAULT_TIMEWAITSYNCMODE)
50 0 : {}
51 :
52 19 : SendRecvExecutor::SendRecvExecutor(HcclRtStream stream, QpHandle qpHandle, AscendSendRecvLinkInfo* linkInfo)
53 19 : : stream_(stream),
54 19 : qpHandle_(qpHandle),
55 19 : localSyncMemPrepare_(AscendMrInfo2MrInfo(linkInfo->localSyncMemPrepare)),
56 19 : localSyncMemDone_(AscendMrInfo2MrInfo(linkInfo->localSyncMemDone)),
57 19 : localSyncMemAck_(AscendMrInfo2MrInfo(linkInfo->localSyncMemAck)),
58 19 : remoteSyncMemPrepare_(AscendMrInfo2MrInfo(linkInfo->remoteSyncMemPrepare)),
59 19 : remoteSyncMemDone_(AscendMrInfo2MrInfo(linkInfo->remoteSyncMemDone)),
60 19 : remoteSyncMemAck_(AscendMrInfo2MrInfo(linkInfo->remoteSyncMemAck)),
61 19 : immData_(linkInfo->immData),
62 19 : wqePerDoorBell_(linkInfo->wqePerDoorbell)
63 19 : {}
64 :
65 10 : SendRecvExecutor::SendRecvExecutor(HcclRtStream stream, QpHandle qpHandle, AscendSendLinkInfo* linkInfo)
66 10 : : stream_(stream),
67 10 : qpHandle_(qpHandle),
68 10 : localSyncMemAck_(AscendMrInfo2MrInfo(linkInfo->localSyncMemAck)),
69 10 : wqePerDoorBell_(linkInfo->wqePerDoorbell),
70 10 : remoteNotifyValueMem_(AscendMrInfo2MrInfo(linkInfo->remoteNotifyValueMem))
71 10 : {}
72 :
73 19 : SendRecvExecutor::SendRecvExecutor(
74 19 : HcclRtStream stream, QpHandle qpHandle, AscendMrInfo* localSyncMemDone, AscendMrInfo* remoteSyncMemAck)
75 19 : : stream_(stream),
76 19 : qpHandle_(qpHandle),
77 19 : localSyncMemDone_(AscendMrInfo2MrInfo(localSyncMemDone)),
78 19 : remoteSyncMemAck_(AscendMrInfo2MrInfo(remoteSyncMemAck))
79 19 : {}
80 :
81 48 : SendRecvExecutor::~SendRecvExecutor() {}
82 :
83 19 : HcclResult SendRecvExecutor::Init()
84 : {
85 19 : CHK_RET(hrtGetNotifySize(notifySize_));
86 19 : if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET) {
87 0 : notifyWaitMode_ = SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE;
88 : }
89 19 : CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(
90 : reinterpret_cast<u64>(localSyncMemPrepare_.addr), prepareNotify_));
91 19 : CHK_PTR_NULL(prepareNotify_);
92 :
93 19 : CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(reinterpret_cast<u64>(localSyncMemAck_.addr), ackNotify_));
94 19 : CHK_PTR_NULL(ackNotify_);
95 :
96 19 : CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(reinterpret_cast<u64>(localSyncMemDone_.addr), doneNotify_));
97 19 : CHK_PTR_NULL(doneNotify_);
98 :
99 19 : CHK_RET(TypicalSyncMem::GetInstance().GetNotifySrcMem(notifySrcMem_));
100 19 : CHK_PTR_NULL(notifySrcMem_.addr);
101 :
102 19 : HCCL_INFO(
103 : "[SendRecvExecutor][Init] SendRecvExecutor init success! notifySize[%u], notifyWaitMode[%d], "
104 : "prepareNotify[%p], ackNotify[%p], doneNotify[%p], notifySrcMem addr[%p], localWindowMem addr[%p], "
105 : "remoteWindowMem addr[%p], remoteSyncMemPrepare addr[%p], remoteSyncMemDone addr[%p], "
106 : "remoteSyncMemAck addr[%p], immData[%u], wqePerDoorBell[%u]",
107 : notifySize_, notifyWaitMode_, prepareNotify_, ackNotify_, doneNotify_, notifySrcMem_.addr, localWindowMem_.addr,
108 : remoteWindowMem_.addr, remoteSyncMemPrepare_.addr, remoteSyncMemDone_.addr, remoteSyncMemAck_.addr, immData_,
109 : wqePerDoorBell_);
110 19 : return HCCL_SUCCESS;
111 : }
112 :
113 19 : HcclResult SendRecvExecutor::WaitPutInit()
114 : {
115 19 : CHK_RET(hrtGetNotifySize(notifySize_));
116 19 : if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET) {
117 0 : notifyWaitMode_ = SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE;
118 : }
119 19 : CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(reinterpret_cast<u64>(localSyncMemDone_.addr), doneNotify_));
120 19 : CHK_PTR_NULL(doneNotify_);
121 :
122 19 : CHK_RET(TypicalSyncMem::GetInstance().GetNotifySrcMem(notifySrcMem_));
123 19 : CHK_PTR_NULL(notifySrcMem_.addr);
124 :
125 19 : HCCL_INFO(
126 : "[SendRecvExecutor][WaitPutInit] SendRecvExecutor init success! notifySize[%u], notifyWaitMode[%d], "
127 : "prepareNotify[%p], ackNotify[%p], doneNotify[%p], notifySrcMem addr[%p], localWindowMem addr[%p], "
128 : "remoteWindowMem addr[%p], remoteSyncMemPrepare addr[%p], remoteSyncMemDone addr[%p], "
129 : "remoteSyncMemAck addr[%p], immData[%u], wqePerDoorBell[%u]",
130 : notifySize_, notifyWaitMode_, prepareNotify_, ackNotify_, doneNotify_, notifySrcMem_.addr, localWindowMem_.addr,
131 : remoteWindowMem_.addr, remoteSyncMemPrepare_.addr, remoteSyncMemDone_.addr, remoteSyncMemAck_.addr, immData_,
132 : wqePerDoorBell_);
133 19 : return HCCL_SUCCESS;
134 : }
135 :
136 0 : HcclResult SendRecvExecutor::IsOverlappedWithWinMem(void* userPtr, u64 userMemSize, bool& isOverlapped)
137 : {
138 0 : if (userPtr >= localWindowMem_.addr
139 0 : && static_cast<u8*>(userPtr) + userMemSize <= static_cast<u8*>(localWindowMem_.addr) + localWindowMem_.size) {
140 0 : isOverlapped = true;
141 0 : } else if (
142 0 : static_cast<u8*>(userPtr) + userMemSize <= static_cast<u8*>(localWindowMem_.addr)
143 0 : || static_cast<u8*>(userPtr) >= static_cast<u8*>(localWindowMem_.addr) + localWindowMem_.size) {
144 0 : isOverlapped = false;
145 : } else {
146 0 : HCCL_ERROR(
147 : "[SendRecvExecutor][IsOverlappedWithWinMem] The user mem addr or size is illegal. "
148 : "The addr of user mem is %p, user mem size is %llu. The addr of window mem is %p, window mem size is %llu.",
149 : userPtr, userMemSize, localWindowMem_.addr, localWindowMem_.size);
150 0 : return HCCL_E_PARA;
151 : }
152 0 : return HCCL_SUCCESS;
153 : }
154 :
155 0 : HcclResult SendRecvExecutor::Send(void* inputPtr, u64 count, HcclDataType dataType)
156 : {
157 0 : HcclResult ret = HCCL_SUCCESS;
158 0 : void* windowsMemPtr = localWindowMem_.addr;
159 0 : uint32_t unitSize = SIZE_TABLE[dataType];
160 0 : uint8_t* curInputPtr = static_cast<uint8_t*>(inputPtr);
161 0 : CHK_PTR_NULL(curInputPtr);
162 0 : uint64_t inputOffset = 0;
163 0 : uint64_t countLeft = count;
164 0 : Stream streamObj(stream_);
165 : // 判断userMem是否是windowMem的一部分
166 0 : bool isOverlapped = false;
167 0 : CHK_RET(IsOverlappedWithWinMem(inputPtr, count * unitSize, isOverlapped));
168 :
169 0 : u64 maxCountPerLoop = localWindowMem_.size / unitSize;
170 :
171 0 : while (countLeft > 0) {
172 : // 防止数据回绕
173 0 : CHK_PRT_RET(countLeft > count, HCCL_ERROR("[SendRecvExecutor][Send] countLeft is underflow."), HCCL_E_PARA);
174 0 : curInputPtr += inputOffset;
175 0 : HCCL_DEBUG("[SendRecvExecutor]][Send] InputOffset[%llu]", inputOffset);
176 0 : u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
177 0 : u64 curSize = curCount * unitSize;
178 0 : HCCL_DEBUG(
179 : "[SendRecvExecutor][Send] curInputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr, curCount, curSize);
180 0 : DeviceMem inMem(curInputPtr, curSize);
181 : // 如果userMem是否是windowMem的一部分，跳过D2D拷
182 0 : if (isOverlapped) {
183 0 : ret = SendRun(inMem);
184 : } else {
185 0 : DeviceMem inWindowMem(windowsMemPtr, curSize);
186 0 : CHK_RET(MemcpyAsyncD2D(inWindowMem, inMem, streamObj));
187 0 : ret = SendRun(inWindowMem);
188 0 : }
189 0 : CHK_PRT_RET(
190 : ret != HCCL_SUCCESS,
191 : HCCL_ERROR(
192 : "[SendRecvExecutor][Send] errNo[0x%016llx] send error, ptr[%p], count[%llu], dataType[%d]",
193 : HCCL_ERROR_CODE(ret), windowsMemPtr, curCount, dataType),
194 : ret);
195 0 : CHK_PRT_RET((curCount == 0), HCCL_ERROR("[SendRecvExecutor]][Send]In OP_BASE curCount is zero"), HCCL_E_PARA);
196 0 : countLeft -= curCount;
197 0 : inputOffset = curSize;
198 0 : }
199 0 : return HCCL_SUCCESS;
200 0 : }
201 :
202 0 : HcclResult SendRecvExecutor::Receive(void* outputPtr, u64 count, HcclDataType dataType)
203 : {
204 0 : HcclResult ret = HCCL_SUCCESS;
205 0 : void* windowsMemPtr = localWindowMem_.addr;
206 0 : uint32_t unitSize = SIZE_TABLE[dataType];
207 0 : uint8_t* curOutPutPtr = static_cast<uint8_t*>(outputPtr);
208 0 : CHK_PTR_NULL(curOutPutPtr);
209 0 : uint64_t outputOffset = 0;
210 0 : uint64_t countLeft = count;
211 0 : Stream streamObj(stream_);
212 0 : u64 maxCountPerLoop = localWindowMem_.size / unitSize;
213 :
214 0 : while (countLeft > 0) {
215 : // 防止数据回绕
216 0 : CHK_PRT_RET(countLeft > count, HCCL_ERROR("[SendRecvExecutor][Receive] countLeft is underflow."), HCCL_E_PARA);
217 0 : curOutPutPtr += outputOffset;
218 0 : HCCL_INFO("[SendRecvExecutor][Receive] inputOffset[%llu]", outputOffset);
219 0 : u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
220 0 : u64 curSize = curCount * unitSize; // 单位 byte
221 0 : HCCL_INFO(
222 : "[SendRecvExecutor][Receive] curOutPutPtr[%p], curCount[%llu], curSize[%llu]", curOutPutPtr, curCount,
223 : curSize);
224 0 : DeviceMem outMem(curOutPutPtr, curSize);
225 0 : DeviceMem outWindowMem(windowsMemPtr, curSize);
226 0 : if (immData_ != 0) {
227 0 : ret = ReceiveRunByPollCq(outWindowMem);
228 : } else {
229 0 : ret = ReceiveRun(outWindowMem);
230 : }
231 0 : CHK_PRT_RET(
232 : ret != HCCL_SUCCESS,
233 : HCCL_ERROR(
234 : "[SendRecvExecutor][Receive] errNo[0x%016llx] Receive error, ptr[%p], count[%llu], dataType[%d]",
235 : HCCL_ERROR_CODE(ret), windowsMemPtr, curCount, dataType),
236 : ret);
237 0 : CHK_RET(MemcpyAsyncD2D(outMem, outWindowMem, streamObj));
238 0 : CHK_PRT_RET((curCount == 0), HCCL_ERROR("[SendRecvExecutor][Receive]In OP_BASE curCount is zero"), HCCL_E_PARA);
239 0 : countLeft -= curCount;
240 0 : outputOffset = curSize;
241 0 : }
242 0 : return HCCL_SUCCESS;
243 0 : }
244 :
245 0 : HcclResult SendRecvExecutor::Put(void* inputPtr, u64 count, HcclDataType dataType)
246 : {
247 0 : HcclResult ret = HCCL_SUCCESS;
248 0 : void* windowsMemPtr = localWindowMem_.addr;
249 0 : uint32_t unitSize = SIZE_TABLE[dataType];
250 0 : uint8_t* curInputPtr = static_cast<uint8_t*>(inputPtr);
251 0 : CHK_PTR_NULL(curInputPtr);
252 0 : uint64_t inputOffset = 0;
253 0 : uint64_t countLeft = count;
254 0 : Stream streamObj(stream_);
255 : // 判断userMem是否是windowMem的一部分
256 0 : bool isOverlapped = false;
257 0 : CHK_RET(IsOverlappedWithWinMem(inputPtr, count * unitSize, isOverlapped));
258 :
259 0 : u64 maxCountPerLoop = localWindowMem_.size / unitSize;
260 :
261 0 : while (countLeft > 0) {
262 : // 防止数据回绕
263 0 : CHK_PRT_RET(countLeft > count, HCCL_ERROR("[SendRecvExecutor][Put] countLeft is underflow."), HCCL_E_PARA);
264 0 : curInputPtr += inputOffset;
265 0 : HCCL_DEBUG("[SendRecvExecutor][Put] InputOffset[%llu]", inputOffset);
266 0 : u64 curCount = (countLeft > maxCountPerLoop) ? maxCountPerLoop : countLeft;
267 0 : u64 curSize = curCount * unitSize;
268 0 : HCCL_DEBUG(
269 : "[SendRecvExecutor][Put] curInputPtr[%p], curCount[%llu], curSize[%llu]", curInputPtr, curCount, curSize);
270 0 : DeviceMem inMem(curInputPtr, curSize);
271 : // 如果userMem是否是windowMem的一部分，跳过D2D拷贝
272 0 : if (isOverlapped) {
273 0 : ret = PutRun(inMem);
274 : } else {
275 0 : DeviceMem inWindowMem(windowsMemPtr, curSize);
276 0 : CHK_RET(MemcpyAsyncD2D(inWindowMem, inMem, streamObj));
277 0 : ret = PutRun(inWindowMem);
278 0 : }
279 0 : CHK_PRT_RET(
280 : ret != HCCL_SUCCESS,
281 : HCCL_ERROR(
282 : "[SendRecvExecutor][Put] errNo[0x%016llx] send error, ptr[%p], count[%llu], dataType[%d]",
283 : HCCL_ERROR_CODE(ret), windowsMemPtr, curCount, dataType),
284 : ret);
285 0 : CHK_PRT_RET((curCount == 0), HCCL_ERROR("[SendRecvExecutor]][Put]In OP_BASE curCount is zero"), HCCL_E_PARA);
286 0 : countLeft -= curCount;
287 0 : inputOffset = curSize;
288 0 : }
289 0 : return HCCL_SUCCESS;
290 0 : }
291 :
292 0 : HcclResult SendRecvExecutor::SendRun(DeviceMem& sendBuffer)
293 : {
294 0 : HcclResult ret = HCCL_SUCCESS;
295 0 : if (!sendBuffer) {
296 0 : HCCL_ERROR("[SendRecvExecutor][SendRun] Send buffer ptr is null.");
297 0 : return HCCL_E_PTR;
298 : }
299 0 : u64 sizePerRound = 0;
300 0 : u64 sizePerSlice = chunkSize_;
301 0 : u64 length = sendBuffer.size();
302 0 : u64 offset = 0;
303 :
304 0 : for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
305 : // 防止数据回绕
306 0 : CHK_PRT_RET(
307 : sizeResidue > length, HCCL_ERROR("[SendRecvExecutor][SendRun] countLeft is underflow."), HCCL_E_PARA);
308 0 : ret = WaitSignal(prepareNotify_);
309 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][SendRun] Wait prepare failed"), ret);
310 0 : offset += sizePerRound;
311 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
312 0 : void* localAddr = static_cast<u8*>(localWindowMem_.addr) + offset;
313 0 : HCCL_INFO("rx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);
314 :
315 0 : ret = PayLoad(localAddr, offset, sizePerRound);
316 0 : CHK_PRT_RET(
317 : ret != HCCL_SUCCESS,
318 : HCCL_ERROR(
319 : "[SendRecvExecutor][SendRun] Send data fail with offset[%llu] "
320 : "size[%llu] failed",
321 : offset, sizePerRound),
322 : ret);
323 :
324 0 : if (immData_ == 0) {
325 0 : ret = RecordNotify(
326 0 : remoteSyncMemDone_.addr, remoteSyncMemDone_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
327 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][SendRun] Record done failed"), ret);
328 : }
329 :
330 0 : ret = WaitSignal(ackNotify_);
331 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][SendRun] Wait ack failed"), ret);
332 : }
333 0 : return HCCL_SUCCESS;
334 : }
335 :
336 0 : HcclResult SendRecvExecutor::PutRun(DeviceMem& putBuffer)
337 : {
338 0 : HcclResult ret = HCCL_SUCCESS;
339 0 : if (!putBuffer) {
340 0 : HCCL_ERROR("[SendRecvExecutor][PutRun] Send buffer ptr is null.");
341 0 : return HCCL_E_PTR;
342 : }
343 0 : u64 sizePerRound = 0;
344 0 : u64 sizePerSlice = chunkSize_;
345 0 : u64 length = putBuffer.size();
346 0 : u64 offset = 0;
347 :
348 0 : for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
349 : // 防止数据回绕
350 0 : CHK_PRT_RET(
351 : sizeResidue > length, HCCL_ERROR("[SendRecvExecutor][PutRun] countLeft is underflow."), HCCL_E_PARA);
352 0 : offset += sizePerRound;
353 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
354 0 : void* localAddr = static_cast<u8*>(localWindowMem_.addr) + offset;
355 0 : HCCL_INFO("rx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);
356 :
357 0 : ret = PayLoad(localAddr, offset, sizePerRound);
358 0 : CHK_PRT_RET(
359 : ret != HCCL_SUCCESS,
360 : HCCL_ERROR(
361 : "[SendRecvExecutor][PutRun] Send data fail with offset[%llu] "
362 : "size[%llu] failed",
363 : offset, sizePerRound),
364 : ret);
365 :
366 0 : if (immData_ == 0) {
367 0 : ret = RecordNotify(
368 0 : remoteSyncMemDone_.addr, remoteSyncMemDone_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
369 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][PutRun] Record done failed"), ret);
370 : }
371 :
372 0 : ret = WaitSignal(ackNotify_);
373 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][PutRun] Wait ack failed"), ret);
374 : }
375 0 : return HCCL_SUCCESS;
376 : }
377 :
378 0 : HcclResult SendRecvExecutor::PollCq()
379 : {
380 0 : auto startTime = std::chrono::steady_clock::now();
381 0 : auto timeout = std::chrono::seconds(static_cast<s32>(GetExternalInputHcclExecTimeOut()));
382 :
383 : struct ibv_wc wc[HCCL_POLL_CQ_DEPTH];
384 0 : while ((std::chrono::steady_clock::now() - startTime) < timeout) {
385 0 : s32 num = hrtRaPollCq(qpHandle_, false, HCCL_POLL_CQ_ONETIME, wc);
386 0 : if (num < 0) {
387 0 : HCCL_ERROR("[SendRecvExecutor][PollCq] Poll Cq error, return [%d]", num);
388 0 : return HCCL_E_INTERNAL;
389 0 : } else if (num < HCCL_POLL_CQ_ONETIME) {
390 0 : SaluSleep(HCCL_POLL_CQ_INTERVAL);
391 0 : continue;
392 : }
393 0 : for (int i = 0; i < num; i++) {
394 0 : if (wc[i].status != 0) {
395 0 : HCCL_ERROR("rdma poll tag sq failed, cqe status[%u]", wc[i].status);
396 0 : return HCCL_E_INTERNAL;
397 : }
398 : }
399 0 : return HCCL_SUCCESS;
400 : }
401 0 : HCCL_ERROR("[SendRecvExecutor][PollCq] Wait Cqe timeOut[%d] s", GetExternalInputHcclLinkTimeOut());
402 0 : return HCCL_E_TIMEOUT;
403 : }
404 :
405 0 : HcclResult SendRecvExecutor::ReceiveRunByPollCq(DeviceMem& receiveBuffer)
406 : {
407 0 : HcclResult ret = HCCL_SUCCESS;
408 0 : if (!receiveBuffer) {
409 0 : HCCL_ERROR("[SendRecvExecutor][ReceiveRunByPollCq] Receive buffer ptr is null.");
410 0 : return HCCL_E_PTR;
411 : }
412 0 : u64 sizePerRound = 0;
413 0 : u64 sizePerSlice = chunkSize_;
414 0 : u64 length = receiveBuffer.size();
415 :
416 0 : u64 offset = 0;
417 :
418 0 : for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
419 : // 防止数据回绕
420 0 : CHK_PRT_RET(
421 : sizeResidue > length, HCCL_ERROR("[SendRecvExecutor][ReceiveRunByPollCq] countLeft is underflow."),
422 : HCCL_E_PARA);
423 0 : offset += sizePerRound;
424 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
425 0 : HCCL_INFO("rx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);
426 :
427 0 : void* localAddr = static_cast<u8*>(localWindowMem_.addr) + offset;
428 :
429 0 : std::vector<struct RecvWrlistData> recvWrVec(1);
430 0 : recvWrVec[0].wrId = reinterpret_cast<u64>(localWindowMem_.addr);
431 0 : recvWrVec[0].memList.addr = reinterpret_cast<u64>(localAddr);
432 0 : recvWrVec[0].memList.len = sizePerRound;
433 0 : recvWrVec[0].memList.lkey = localWindowMem_.lkey;
434 :
435 0 : struct RecvWrlistData* recvWr = recvWrVec.data();
436 0 : u32 completeNum = 0;
437 0 : ret = hrtRaRecvWrlist(qpHandle_, recvWr, 1, &completeNum);
438 0 : if (ret == HCCL_SUCCESS && completeNum == 1) {
439 0 : HCCL_INFO("[SendRecvExecutor][ReceiveRunByPollCq] Exec hrtRaRecvWrlist success.");
440 : } else {
441 0 : HCCL_ERROR(
442 : "[SendRecvExecutor][ReceiveRunByPollCq] In RdmaDataTransport, hrtRaRecvWrlist failed. ret[%d], "
443 : "completeNum[%d].",
444 : ret, completeNum);
445 0 : return HCCL_E_NETWORK;
446 : }
447 :
448 0 : ret = RecordNotify(
449 0 : remoteSyncMemPrepare_.addr, remoteSyncMemPrepare_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey,
450 0 : notifySize_);
451 0 : CHK_PRT_RET(
452 : ret != HCCL_SUCCESS, HCCL_ERROR("SendRecvExecutor][ReceiveRunByPollCq] Record prepare failed"), ret);
453 :
454 0 : CHK_RET(PollCq());
455 :
456 0 : ret = RecordNotify(
457 0 : remoteSyncMemAck_.addr, remoteSyncMemAck_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
458 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][ReceiveRunByPollCq] Record ack failed"), ret);
459 0 : }
460 0 : return HCCL_SUCCESS;
461 : }
462 :
463 0 : HcclResult SendRecvExecutor::ReceiveRun(DeviceMem& receiveBuffer)
464 : {
465 0 : HcclResult ret = HCCL_SUCCESS;
466 0 : if (!receiveBuffer) {
467 0 : HCCL_ERROR("[SendRecvExecutor][ReceiveRun] Receive buffer ptr is null.");
468 0 : return HCCL_E_PTR;
469 : }
470 0 : u64 sizePerRound = 0;
471 0 : u64 sizePerSlice = chunkSize_;
472 0 : u64 length = receiveBuffer.size();
473 0 : u64 offset = 0;
474 :
475 0 : for (u64 sizeResidue = length; sizeResidue > 0; sizeResidue -= sizePerRound) {
476 : // 防止数据回绕
477 0 : CHK_PRT_RET(
478 : sizeResidue > length, HCCL_ERROR("[SendRecvExecutor][ReceiveRun] countLeft is underflow."), HCCL_E_PARA);
479 0 : ret = RecordNotify(
480 0 : remoteSyncMemPrepare_.addr, remoteSyncMemPrepare_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey,
481 0 : notifySize_);
482 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("SendRecvExecutor][ReceiveRun] Record prepare failed"), ret);
483 0 : offset += sizePerRound;
484 0 : sizePerRound = (sizeResidue > sizePerSlice) ? sizePerSlice : sizeResidue;
485 :
486 0 : HCCL_INFO("[SendRecvExecutor][ReceiveRun]rx async inputmem's offset[%llu] size[%llu]", offset, sizePerRound);
487 :
488 0 : ret = WaitSignal(doneNotify_);
489 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][ReceiveRun] Wait done failed"), ret);
490 :
491 0 : ret = RecordNotify(
492 0 : remoteSyncMemAck_.addr, remoteSyncMemAck_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
493 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][ReceiveRun] Wait ack failed"), ret);
494 : }
495 0 : return HCCL_SUCCESS;
496 : }
497 :
498 48 : HcclResult SendRecvExecutor::RecordNotify(
499 : void* dstMemPtr, u32 rkey, const void* srcMemPtr, u32 lkey, u64 srcMemSize, uint32_t rdmaOp, int sendFlag)
500 : {
501 48 : struct SgList list = {};
502 48 : struct SendWrV2 wr = {};
503 : // 构造wr信息
504 48 : list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(srcMemPtr));
505 48 : list.len = srcMemSize;
506 48 : list.lkey = lkey;
507 :
508 48 : wr.bufList = &list;
509 48 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
510 48 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(dstMemPtr));
511 48 : wr.rkey = rkey;
512 48 : wr.op = rdmaOp;
513 48 : wr.sendFlag = sendFlag;
514 :
515 48 : HCCL_INFO(
516 : "[SendRecvExecutor][RecordNotify] "
517 : "Notify's dst addr[%p], local addr[%p], data's len[%u], remote mr key[%u], local mr key[%u]",
518 : wr.dstAddr, wr.bufList->addr, wr.bufList->len, wr.rkey, wr.bufList->lkey);
519 :
520 : // RDMA异步发送
521 48 : CHK_RET(RdmaSendAsync(wr));
522 48 : return HCCL_SUCCESS;
523 : }
524 :
525 29 : HcclResult SendRecvExecutor::WaitSignal(HcclRtSignal signal)
526 : {
527 29 : if (notifyWaitMode_ == SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE) {
528 0 : CHK_RET(
529 : hrtNotifyWaitWithTimeOut(static_cast<HcclRtNotify>(signal), stream_, GetExternalInputHcclExecTimeOut()));
530 : } else {
531 29 : CHK_RET(hrtNotifyWaitWithTimeOut(static_cast<HcclRtNotify>(signal), stream_, NOTIFY_DEFAULT_WAIT_TIME));
532 : }
533 29 : return HCCL_SUCCESS;
534 : }
535 :
536 0 : HcclResult SendRecvExecutor::PayLoad(const void* src, u64 dstOffset, u64 len)
537 : {
538 : HcclResult ret;
539 0 : HCCL_DEBUG(
540 : "[SendRecvExecutor][PayLoad] Local window memory srcPtr[%p] len[%llu] dstOffset[%llu]", src, len, dstOffset);
541 :
542 0 : u32 txSendDataTimes = (len == 0) ? 1 : (len + RDMA_SEND_MAX_SIZE - 1) / RDMA_SEND_MAX_SIZE;
543 :
544 0 : for (u32 txSendDataIdx = 0; txSendDataIdx < txSendDataTimes; txSendDataIdx++) {
545 0 : u64 txSendDataOffset = txSendDataIdx * RDMA_SEND_MAX_SIZE;
546 0 : u64 txSendDataSize = (txSendDataIdx == (txSendDataTimes - 1)) ? len - txSendDataOffset : RDMA_SEND_MAX_SIZE;
547 :
548 0 : void* txdstMemPtr
549 0 : = reinterpret_cast<void*>(reinterpret_cast<u8*>(remoteWindowMem_.addr) + dstOffset + txSendDataOffset);
550 :
551 0 : const void* txsrcMemPtr = reinterpret_cast<const void*>(reinterpret_cast<const char*>(src) + txSendDataOffset);
552 0 : struct SendWrV2 wr {};
553 : // 构造wr信息
554 0 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
555 0 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(txdstMemPtr));
556 0 : wr.rkey = remoteWindowMem_.lkey;
557 0 : wr.sendFlag = RA_SEND_SIGNALED;
558 0 : if (immData_ != 0) {
559 0 : wr.op = RA_WR_RDMA_WRITE_WITH_IMM;
560 0 : wr.ext.immData = immData_;
561 : } else {
562 0 : wr.op = RA_WR_RDMA_WRITE;
563 : }
564 0 : struct SgList list = {};
565 0 : list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(txsrcMemPtr));
566 0 : list.len = txSendDataSize;
567 0 : list.lkey = localWindowMem_.lkey;
568 0 : wr.bufList = &list;
569 0 : ret = RdmaSendAsync(wr);
570 0 : CHK_PRT_RET(
571 : ret != HCCL_SUCCESS,
572 : HCCL_ERROR(
573 : "[SendRecvExecutor][PayLoad]errNo[0x%016llx] In lbv exp, add wqe list failed."
574 : "srcMemSize[%llu]",
575 : HCCL_ERROR_CODE(ret), txSendDataSize),
576 : ret);
577 : }
578 :
579 0 : return HCCL_SUCCESS;
580 : }
581 :
582 48 : HcclResult SendRecvExecutor::RdmaSendAsync(struct SendWrV2& wr)
583 : {
584 48 : HcclResult ret = HCCL_SUCCESS;
585 48 : struct SendWrRsp opRsp = {};
586 48 : HCCL_DEBUG(
587 : "[SendRecvExecutor][RdmaSendAsync] dst_addr[%p], src_addr[%p], len[%u]", wr.dstAddr, wr.bufList->addr,
588 : wr.bufList->len);
589 :
590 48 : CHK_RET(HrtRaSendWrV2(qpHandle_, &wr, &opRsp, HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));
591 :
592 48 : u32 dbIndex = static_cast<u32>(opRsp.db.dbIndex);
593 48 : u64 dbInfo = static_cast<u64>(opRsp.db.dbInfo);
594 :
595 48 : if ((dbIndex == INVALID_UINT) && (dbInfo == INVALID_U64)) {
596 : // zero byte message 不需要下发rdma send task
597 0 : HCCL_DEBUG("[SendRecvExecutor][RdmaSendAsync] dbIndex and dbInfo is invalid.");
598 0 : return HCCL_SUCCESS;
599 : }
600 :
601 48 : ret = hrtRDMADBSend(dbIndex, dbInfo, stream_);
602 48 : CHK_PRT_RET(
603 : ret != HCCL_SUCCESS,
604 : HCCL_ERROR(
605 : "[SendRecvExecutor][RdmaSendAsync]errNo[0x%016llx] In lbv exp op base mode, "
606 : "rdma send failed. dbIndex[%u] dbInfo[%llu]",
607 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo),
608 : ret);
609 48 : return HCCL_SUCCESS;
610 : }
611 :
612 0 : HcclResult SendRecvExecutor::MemcpyAsyncD2D(hccl::DeviceMem& dst, const hccl::DeviceMem& src, hccl::Stream& stream)
613 : {
614 0 : CHK_PTR_NULL(dst.ptr());
615 0 : CHK_PTR_NULL(src.ptr());
616 :
617 0 : if (stream.ptr() == nullptr) {
618 0 : CHK_SAFETY_FUNC_RET(memcpy_s(dst.ptr(), dst.size(), src.ptr(), src.size()));
619 0 : return HCCL_E_PARA;
620 : }
621 :
622 0 : if (src.size() == 0) {
623 0 : HCCL_DEBUG("[SendRecvExecutor][MemcpyAsyncD2D] count is 0, return success.");
624 0 : return HCCL_SUCCESS;
625 : }
626 :
627 0 : uint64_t spiltLoop = 0;
628 0 : uint64_t addrOffset = 0;
629 0 : uint64_t contSplit = 0;
630 0 : if (src.size() > HCCL_SDMA_MAX_COUNT_4GB) {
631 0 : spiltLoop = (src.size() % HCCL_SDMA_MAX_COUNT_4GB) ? (src.size() / HCCL_SDMA_MAX_COUNT_4GB) :
632 0 : ((src.size() / HCCL_SDMA_MAX_COUNT_4GB) - 1);
633 0 : HCCL_INFO(
634 : "[SendRecvExecutor][MemcpyAsyncD2D] MemcpyAsync SDMA task countSize is bigger than 4GB "
635 : "and do segmentation splitloop[%llu]",
636 : spiltLoop);
637 : }
638 : /* SDMA任务拆分 */
639 0 : for (uint64_t index = 0; index <= spiltLoop; index++) {
640 0 : addrOffset = index * HCCL_SDMA_MAX_COUNT_4GB;
641 0 : contSplit = (index == spiltLoop) ? (src.size() - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
642 0 : void* srcSplit = static_cast<void*>(static_cast<u8*>(const_cast<void*>(src.ptr())) + addrOffset);
643 0 : void* dstSplit = static_cast<void*>(static_cast<u8*>(dst.ptr()) + addrOffset);
644 :
645 0 : CHK_RET(hrtMemAsyncCopy(
646 : dstSplit, dst.size(), const_cast<const void*>(srcSplit), contSplit,
647 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
648 : }
649 0 : return HCCL_SUCCESS;
650 : }
651 :
652 19 : HcclResult SendRecvExecutor::BatchPutMR(u32 num, AscendMrInfo* putMRList, AscendMrInfo* remoteMRList)
653 : {
654 19 : HcclResult ret = HCCL_SUCCESS;
655 19 : u32 sendWrNum = 0;
656 38 : for (u32 i = 0; i < num; i++) {
657 19 : if (i != num - 1) {
658 0 : CHK_RET(PayLoadMR(putMRList + i, remoteMRList + i, sendWrNum));
659 : } else {
660 : // 最后一组数据特殊处理，发送立即数
661 19 : CHK_RET(PayLoadMR(putMRList + i, remoteMRList + i, sendWrNum, true));
662 : }
663 : }
664 :
665 19 : if (immData_ == 0) {
666 38 : ret = RecordNotify(
667 19 : remoteSyncMemDone_.addr, remoteSyncMemDone_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
668 19 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][BatchPutMR] Record done failed"), ret);
669 : }
670 :
671 19 : ret = WaitSignal(ackNotify_);
672 19 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][BatchPutMR] Wait ack failed"), ret);
673 :
674 19 : return HCCL_SUCCESS;
675 : }
676 :
677 : HcclResult
678 29 : SendRecvExecutor::PayLoadMR(AscendMrInfo* putMRInfo, AscendMrInfo* remoteMRInfo, u32& wrNum, bool isLastMRtoPut)
679 : {
680 29 : CHK_PRT_RET(
681 : putMRInfo->size != remoteMRInfo->size,
682 : HCCL_ERROR(
683 : "[SendRecvExecutor][PayLoadMR] The size of localMR"
684 : "is different from remoteMR. LocalMR size is [%u], remoteMR size is[%u].",
685 : putMRInfo->size, remoteMRInfo->size),
686 : HCCL_E_PARA);
687 29 : u64 remainingSize = putMRInfo->size;
688 29 : u64 offSet = 0;
689 29 : u64 byteChunkSize = 0;
690 29 : u64 localAddr = 0;
691 29 : u64 remoteAddr = 0;
692 29 : bool isLastSlice = false;
693 58 : while (remainingSize > 0) {
694 29 : localAddr = putMRInfo->addr + offSet;
695 29 : remoteAddr = remoteMRInfo->addr + offSet;
696 29 : byteChunkSize = remainingSize > MAX_RDMA_WQE_SIZE ? MAX_RDMA_WQE_SIZE : remainingSize;
697 29 : isLastSlice = remainingSize > MAX_RDMA_WQE_SIZE ? false : true;
698 29 : struct SgList list = {};
699 29 : list.addr = localAddr;
700 29 : list.len = byteChunkSize;
701 29 : list.lkey = putMRInfo->key;
702 29 : struct SendWrV2 wr = {};
703 29 : wr.bufList = &list;
704 29 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
705 29 : wr.dstAddr = remoteAddr;
706 29 : wr.rkey = remoteMRInfo->key;
707 29 : if (isLastMRtoPut && isLastSlice && immData_ != 0) {
708 0 : wr.op = RA_WR_RDMA_WRITE_WITH_IMM;
709 0 : wr.ext.immData = immData_;
710 : } else {
711 29 : wr.op = RA_WR_RDMA_WRITE;
712 : }
713 :
714 29 : CHK_RET(MultiWqeOneDoorBellSend(isLastMRtoPut && isLastSlice, wrNum, wr));
715 29 : remainingSize -= byteChunkSize;
716 29 : offSet += byteChunkSize;
717 : }
718 29 : return HCCL_SUCCESS;
719 : }
720 :
721 29 : HcclResult SendRecvExecutor::MultiWqeOneDoorBellSend(bool isLastWr, u32& wrNum, struct SendWrV2& wr)
722 : {
723 29 : HcclResult ret = HCCL_SUCCESS;
724 29 : wrNum++;
725 : // 多个wqe生成一个cqe
726 29 : if (isLastWr || wrNum == wqePerDoorBell_) {
727 29 : wr.sendFlag = RA_SEND_SIGNALED;
728 : } else {
729 0 : wr.sendFlag = 0;
730 : }
731 29 : struct SendWrRsp opRsp = {};
732 29 : CHK_RET(HrtRaSendWrV2(qpHandle_, &wr, &opRsp, HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));
733 29 : HCCL_DEBUG(
734 : "[SendRecvExecutor][MultiWqeOneDoorBellSend] End SendWr, wr op[%d], localAddr[%p], remoteAddr[%p], "
735 : "len[%llu], local key[%u], remote key[%u].",
736 : wr.op, wr.bufList->addr, wr.dstAddr, wr.bufList->len, wr.bufList->lkey, wr.rkey);
737 :
738 29 : if (static_cast<u32>(opRsp.db.dbIndex) == INVALID_UINT && static_cast<u64>(opRsp.db.dbInfo) == INVALID_U64) {
739 : // zero byte message 不需要下发rdma send task
740 0 : HCCL_DEBUG("[SendRecvExecutor][MultiWqeOneDoorBellSend] dbIndex and dbInfo is invalid.");
741 0 : return HCCL_SUCCESS;
742 : }
743 :
744 : // 每下发wqePerDoorBell_个wr敲一次doorbell
745 29 : if (isLastWr || wrNum == wqePerDoorBell_) {
746 29 : u32 dbIndex = static_cast<u32>(opRsp.db.dbIndex);
747 29 : u64 dbInfo = static_cast<u64>(opRsp.db.dbInfo);
748 29 : HCCL_DEBUG(
749 : "[SendRecvExecutor][MultiWqeOneDoorBellSend] Start RDMADBSend, dbIndex[%u], dbInfo[%llu], wrNum[%u], "
750 : "isLastMR[%d].",
751 : dbIndex, dbInfo, wrNum, isLastWr);
752 29 : ret = hrtRDMADBSend(dbIndex, dbInfo, stream_);
753 29 : CHK_PRT_RET(
754 : ret != HCCL_SUCCESS,
755 : HCCL_ERROR(
756 : "[SendRecvExecutor][MultiWqeOneDoorBellSend]errNo[0x%016llx] In lbv exp op base mode, "
757 : "rdma send failed. dbIndex[%u] dbInfo[%llu]",
758 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo),
759 : ret);
760 29 : wrNum = 0;
761 : }
762 29 : return HCCL_SUCCESS;
763 : }
764 :
765 19 : HcclResult SendRecvExecutor::WaitPutMR()
766 : {
767 19 : HcclResult ret = HCCL_SUCCESS;
768 :
769 19 : ret = WaitSignalUnlimitedTime(doneNotify_);
770 19 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][WaitPutMR] Wait done failed"), ret);
771 :
772 38 : ret = RecordNotify(
773 19 : remoteSyncMemAck_.addr, remoteSyncMemAck_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
774 19 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][WaitPutMR] Record ack failed"), ret);
775 :
776 19 : return HCCL_SUCCESS;
777 : }
778 :
779 0 : HcclResult SendRecvExecutor::ProcessRCQ(AscendMrInfo* lastMRInfo)
780 : {
781 0 : HcclResult ret = HCCL_SUCCESS;
782 0 : u64 remainingSize = lastMRInfo->size;
783 0 : u64 localAddr = lastMRInfo->addr;
784 0 : u64 offSet = 0;
785 0 : while (remainingSize > MAX_RDMA_WQE_SIZE) {
786 0 : remainingSize -= MAX_RDMA_WQE_SIZE;
787 0 : offSet += MAX_RDMA_WQE_SIZE;
788 : }
789 0 : std::vector<struct RecvWrlistData> recvWrVec(1);
790 0 : recvWrVec[0].wrId = localAddr;
791 0 : recvWrVec[0].memList.addr = localAddr + offSet;
792 0 : recvWrVec[0].memList.len = remainingSize;
793 0 : recvWrVec[0].memList.lkey = lastMRInfo->key;
794 :
795 0 : struct RecvWrlistData* recvWr = recvWrVec.data();
796 0 : u32 completeNum = 0;
797 0 : ret = hrtRaRecvWrlist(qpHandle_, recvWr, 1, &completeNum);
798 0 : if (ret == HCCL_SUCCESS && completeNum == 1) {
799 0 : HCCL_INFO("[SendRecvExecutor][TestBatchPutMR] Exec hrtRaRecvWrlist success.");
800 : } else {
801 0 : HCCL_ERROR(
802 : "[SendRecvExecutor][TestBatchPutMR] In RdmaDataTransport, hrtRaRecvWrlist failed. ret[%d], "
803 : "completeNum[%d].",
804 : ret, completeNum);
805 0 : return HCCL_E_NETWORK;
806 : }
807 :
808 0 : CHK_RET(PollCq());
809 0 : return HCCL_SUCCESS;
810 0 : }
811 :
812 19 : HcclResult SendRecvExecutor::WaitSignalUnlimitedTime(HcclRtSignal signal)
813 : {
814 : // 超时时间配置成0，代表永不超时
815 19 : CHK_RET(hrtNotifyWaitWithTimeOut(static_cast<HcclRtNotify>(signal), stream_, 0));
816 19 : return HCCL_SUCCESS;
817 : }
818 :
819 0 : HcclResult SendRecvExecutor::WaitPutMROnlyWait()
820 : {
821 0 : HcclResult ret = HCCL_SUCCESS;
822 :
823 0 : ret = WaitSignalUnlimitedTime(doneNotify_);
824 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][WaitPutMROnlyWait] Wait done failed"), ret);
825 :
826 0 : return HCCL_SUCCESS;
827 : }
828 :
829 0 : HcclResult SendRecvExecutor::WaitPutMROnlyRecord()
830 : {
831 0 : HcclResult ret = HCCL_SUCCESS;
832 :
833 0 : ret = RecordNotify(
834 0 : remoteSyncMemAck_.addr, remoteSyncMemAck_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
835 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][WaitPutMROnlyWait] Record ack failed"), ret);
836 :
837 0 : return HCCL_SUCCESS;
838 : }
839 :
840 10 : HcclResult SendRecvExecutor::OneSideBatchPutMR(u32 num, AscendMrInfo* putMRList, AscendMrInfo* remoteMRList)
841 : {
842 10 : CHK_RET(hrtGetNotifySize(notifySize_));
843 10 : if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET) {
844 0 : notifyWaitMode_ = SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE;
845 : }
846 10 : CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(reinterpret_cast<u64>(localSyncMemAck_.addr), ackNotify_));
847 10 : CHK_PTR_NULL(ackNotify_);
848 10 : HCCL_INFO(
849 : "[OneSideBatchPutMR] notifySize[%u], notifyWaitMode[%d], "
850 : "ackNotify[%p], remoteNotifyValueMem addr[%p], remoteNotifyValueMem len[%llu], remoteNotifyValueMem key[%u], "
851 : "wqePerDoorBell[%u]",
852 : notifySize_, notifyWaitMode_, ackNotify_, remoteNotifyValueMem_.addr, remoteNotifyValueMem_.size,
853 : remoteNotifyValueMem_.lkey, wqePerDoorBell_);
854 :
855 10 : u32 sendWrNum = 0;
856 20 : for (u32 i = 0; i < num; i++) {
857 10 : if (i != num - 1) {
858 0 : CHK_RET(PayLoadMR(putMRList + i, remoteMRList + i, sendWrNum));
859 : } else {
860 : // 最后一组数据特殊处理，发送立即数
861 10 : CHK_RET(PayLoadMR(putMRList + i, remoteMRList + i, sendWrNum, true));
862 : }
863 : }
864 :
865 20 : HcclResult ret = RecordNotify(
866 10 : remoteNotifyValueMem_.addr, remoteNotifyValueMem_.lkey, localSyncMemAck_.addr, localSyncMemAck_.lkey,
867 10 : notifySize_, RA_WR_RDMA_READ, RA_SEND_SIGNALED | RA_SEND_FENCE);
868 10 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OneSideBatchPutMR] Record ack failed"), ret);
869 :
870 10 : ret = WaitSignal(ackNotify_);
871 10 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OneSideBatchPutMR] Wait ack failed"), ret);
872 :
873 10 : return ret;
874 : }
875 :
876 : } // namespace hccl
|
