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 : // GetExternalInputHcclExecTimeOut 返回 double，接口参数为整型秒数
529 0 : CHK_RET(hrtNotifyWaitWithTimeOut(
530 : static_cast<HcclRtNotify>(signal), stream_, static_cast<s32>(GetExternalInputHcclExecTimeOut())));
531 : } else {
532 29 : CHK_RET(hrtNotifyWaitWithTimeOut(static_cast<HcclRtNotify>(signal), stream_, NOTIFY_DEFAULT_WAIT_TIME));
533 : }
534 29 : return HCCL_SUCCESS;
535 : }
536 :
537 0 : HcclResult SendRecvExecutor::PayLoad(const void* src, u64 dstOffset, u64 len)
538 : {
539 : HcclResult ret;
540 0 : HCCL_DEBUG(
541 : "[SendRecvExecutor][PayLoad] Local window memory srcPtr[%p] len[%llu] dstOffset[%llu]", src, len, dstOffset);
542 :
543 0 : u32 txSendDataTimes = (len == 0) ? 1 : (len + RDMA_SEND_MAX_SIZE - 1) / RDMA_SEND_MAX_SIZE;
544 :
545 0 : for (u32 txSendDataIdx = 0; txSendDataIdx < txSendDataTimes; txSendDataIdx++) {
546 0 : u64 txSendDataOffset = txSendDataIdx * RDMA_SEND_MAX_SIZE;
547 0 : u64 txSendDataSize = (txSendDataIdx == (txSendDataTimes - 1)) ? len - txSendDataOffset : RDMA_SEND_MAX_SIZE;
548 :
549 0 : void* txdstMemPtr
550 0 : = reinterpret_cast<void*>(reinterpret_cast<u8*>(remoteWindowMem_.addr) + dstOffset + txSendDataOffset);
551 :
552 0 : const void* txsrcMemPtr = reinterpret_cast<const void*>(reinterpret_cast<const char*>(src) + txSendDataOffset);
553 0 : struct SendWrV2 wr {};
554 : // 构造wr信息
555 0 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
556 0 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(txdstMemPtr));
557 0 : wr.rkey = remoteWindowMem_.lkey;
558 0 : wr.sendFlag = RA_SEND_SIGNALED;
559 0 : if (immData_ != 0) {
560 0 : wr.op = RA_WR_RDMA_WRITE_WITH_IMM;
561 0 : wr.ext.immData = immData_;
562 : } else {
563 0 : wr.op = RA_WR_RDMA_WRITE;
564 : }
565 0 : struct SgList list = {};
566 0 : list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(txsrcMemPtr));
567 0 : list.len = txSendDataSize;
568 0 : list.lkey = localWindowMem_.lkey;
569 0 : wr.bufList = &list;
570 0 : ret = RdmaSendAsync(wr);
571 0 : CHK_PRT_RET(
572 : ret != HCCL_SUCCESS,
573 : HCCL_ERROR(
574 : "[SendRecvExecutor][PayLoad]errNo[0x%016llx] In lbv exp, add wqe list failed."
575 : "srcMemSize[%llu]",
576 : HCCL_ERROR_CODE(ret), txSendDataSize),
577 : ret);
578 : }
579 :
580 0 : return HCCL_SUCCESS;
581 : }
582 :
583 48 : HcclResult SendRecvExecutor::RdmaSendAsync(struct SendWrV2& wr)
584 : {
585 48 : HcclResult ret = HCCL_SUCCESS;
586 48 : struct SendWrRsp opRsp = {};
587 48 : HCCL_DEBUG(
588 : "[SendRecvExecutor][RdmaSendAsync] dst_addr[%p], src_addr[%p], len[%u]", wr.dstAddr, wr.bufList->addr,
589 : wr.bufList->len);
590 :
591 48 : CHK_RET(HrtRaSendWrV2(qpHandle_, &wr, &opRsp, HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));
592 :
593 48 : u32 dbIndex = static_cast<u32>(opRsp.db.dbIndex);
594 48 : u64 dbInfo = static_cast<u64>(opRsp.db.dbInfo);
595 :
596 48 : if ((dbIndex == INVALID_UINT) && (dbInfo == INVALID_U64)) {
597 : // zero byte message 不需要下发rdma send task
598 0 : HCCL_DEBUG("[SendRecvExecutor][RdmaSendAsync] dbIndex and dbInfo is invalid.");
599 0 : return HCCL_SUCCESS;
600 : }
601 :
602 48 : ret = hrtRDMADBSend(dbIndex, dbInfo, stream_);
603 48 : CHK_PRT_RET(
604 : ret != HCCL_SUCCESS,
605 : HCCL_ERROR(
606 : "[SendRecvExecutor][RdmaSendAsync]errNo[0x%016llx] In lbv exp op base mode, "
607 : "rdma send failed. dbIndex[%u] dbInfo[%llu]",
608 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo),
609 : ret);
610 48 : return HCCL_SUCCESS;
611 : }
612 :
613 0 : HcclResult SendRecvExecutor::MemcpyAsyncD2D(hccl::DeviceMem& dst, const hccl::DeviceMem& src, hccl::Stream& stream)
614 : {
615 0 : CHK_PTR_NULL(dst.ptr());
616 0 : CHK_PTR_NULL(src.ptr());
617 :
618 0 : if (stream.ptr() == nullptr) {
619 0 : CHK_SAFETY_FUNC_RET(memcpy_s(dst.ptr(), dst.size(), src.ptr(), src.size()));
620 0 : return HCCL_E_PARA;
621 : }
622 :
623 0 : if (src.size() == 0) {
624 0 : HCCL_DEBUG("[SendRecvExecutor][MemcpyAsyncD2D] count is 0, return success.");
625 0 : return HCCL_SUCCESS;
626 : }
627 :
628 0 : uint64_t spiltLoop = 0;
629 0 : uint64_t addrOffset = 0;
630 0 : uint64_t contSplit = 0;
631 0 : if (src.size() > HCCL_SDMA_MAX_COUNT_4GB) {
632 0 : spiltLoop = (src.size() % HCCL_SDMA_MAX_COUNT_4GB != 0) ? (src.size() / HCCL_SDMA_MAX_COUNT_4GB) :
633 0 : ((src.size() / HCCL_SDMA_MAX_COUNT_4GB) - 1);
634 0 : HCCL_INFO(
635 : "[SendRecvExecutor][MemcpyAsyncD2D] MemcpyAsync SDMA task countSize is bigger than 4GB "
636 : "and do segmentation splitloop[%llu]",
637 : spiltLoop);
638 : }
639 : /* SDMA任务拆分 */
640 0 : for (uint64_t index = 0; index <= spiltLoop; index++) {
641 0 : addrOffset = index * HCCL_SDMA_MAX_COUNT_4GB;
642 0 : contSplit = (index == spiltLoop) ? (src.size() - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
643 0 : void* srcSplit = static_cast<void*>(static_cast<u8*>(const_cast<void*>(src.ptr())) + addrOffset);
644 0 : void* dstSplit = static_cast<void*>(static_cast<u8*>(dst.ptr()) + addrOffset);
645 :
646 0 : CHK_RET(hrtMemAsyncCopy(
647 : dstSplit, dst.size(), const_cast<const void*>(srcSplit), contSplit,
648 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, stream.ptr()));
649 : }
650 0 : return HCCL_SUCCESS;
651 : }
652 :
653 19 : HcclResult SendRecvExecutor::BatchPutMR(u32 num, AscendMrInfo* putMRList, AscendMrInfo* remoteMRList)
654 : {
655 19 : HcclResult ret = HCCL_SUCCESS;
656 19 : u32 sendWrNum = 0;
657 38 : for (u32 i = 0; i < num; i++) {
658 19 : if (i != num - 1) {
659 0 : CHK_RET(PayLoadMR(putMRList + i, remoteMRList + i, sendWrNum));
660 : } else {
661 : // 最后一组数据特殊处理，发送立即数
662 19 : CHK_RET(PayLoadMR(putMRList + i, remoteMRList + i, sendWrNum, true));
663 : }
664 : }
665 :
666 19 : if (immData_ == 0) {
667 38 : ret = RecordNotify(
668 19 : remoteSyncMemDone_.addr, remoteSyncMemDone_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
669 19 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][BatchPutMR] Record done failed"), ret);
670 : }
671 :
672 19 : ret = WaitSignal(ackNotify_);
673 19 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][BatchPutMR] Wait ack failed"), ret);
674 :
675 19 : return HCCL_SUCCESS;
676 : }
677 :
678 : HcclResult
679 29 : SendRecvExecutor::PayLoadMR(AscendMrInfo* putMRInfo, AscendMrInfo* remoteMRInfo, u32& wrNum, bool isLastMRtoPut)
680 : {
681 29 : CHK_PRT_RET(
682 : putMRInfo->size != remoteMRInfo->size,
683 : HCCL_ERROR(
684 : "[SendRecvExecutor][PayLoadMR] The size of localMR"
685 : "is different from remoteMR. LocalMR size is [%u], remoteMR size is[%u].",
686 : putMRInfo->size, remoteMRInfo->size),
687 : HCCL_E_PARA);
688 29 : u64 remainingSize = putMRInfo->size;
689 29 : u64 offSet = 0;
690 29 : u64 byteChunkSize = 0;
691 29 : u64 localAddr = 0;
692 29 : u64 remoteAddr = 0;
693 29 : bool isLastSlice = false;
694 58 : while (remainingSize > 0) {
695 29 : localAddr = putMRInfo->addr + offSet;
696 29 : remoteAddr = remoteMRInfo->addr + offSet;
697 29 : byteChunkSize = remainingSize > MAX_RDMA_WQE_SIZE ? MAX_RDMA_WQE_SIZE : remainingSize;
698 29 : isLastSlice = remainingSize > MAX_RDMA_WQE_SIZE ? false : true;
699 29 : struct SgList list = {};
700 29 : list.addr = localAddr;
701 29 : list.len = byteChunkSize;
702 29 : list.lkey = putMRInfo->key;
703 29 : struct SendWrV2 wr = {};
704 29 : wr.bufList = &list;
705 29 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
706 29 : wr.dstAddr = remoteAddr;
707 29 : wr.rkey = remoteMRInfo->key;
708 29 : if (isLastMRtoPut && isLastSlice && immData_ != 0) {
709 0 : wr.op = RA_WR_RDMA_WRITE_WITH_IMM;
710 0 : wr.ext.immData = immData_;
711 : } else {
712 29 : wr.op = RA_WR_RDMA_WRITE;
713 : }
714 :
715 29 : CHK_RET(MultiWqeOneDoorBellSend(isLastMRtoPut && isLastSlice, wrNum, wr));
716 29 : remainingSize -= byteChunkSize;
717 29 : offSet += byteChunkSize;
718 : }
719 29 : return HCCL_SUCCESS;
720 : }
721 :
722 29 : HcclResult SendRecvExecutor::MultiWqeOneDoorBellSend(bool isLastWr, u32& wrNum, struct SendWrV2& wr)
723 : {
724 29 : HcclResult ret = HCCL_SUCCESS;
725 29 : wrNum++;
726 : // 多个wqe生成一个cqe
727 29 : if (isLastWr || wrNum == wqePerDoorBell_) {
728 29 : wr.sendFlag = RA_SEND_SIGNALED;
729 : } else {
730 0 : wr.sendFlag = 0;
731 : }
732 29 : struct SendWrRsp opRsp = {};
733 29 : CHK_RET(HrtRaSendWrV2(qpHandle_, &wr, &opRsp, HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE));
734 29 : HCCL_DEBUG(
735 : "[SendRecvExecutor][MultiWqeOneDoorBellSend] End SendWr, wr op[%d], localAddr[%p], remoteAddr[%p], "
736 : "len[%llu], local key[%u], remote key[%u].",
737 : wr.op, wr.bufList->addr, wr.dstAddr, wr.bufList->len, wr.bufList->lkey, wr.rkey);
738 :
739 29 : if (static_cast<u32>(opRsp.db.dbIndex) == INVALID_UINT && static_cast<u64>(opRsp.db.dbInfo) == INVALID_U64) {
740 : // zero byte message 不需要下发rdma send task
741 0 : HCCL_DEBUG("[SendRecvExecutor][MultiWqeOneDoorBellSend] dbIndex and dbInfo is invalid.");
742 0 : return HCCL_SUCCESS;
743 : }
744 :
745 : // 每下发wqePerDoorBell_个wr敲一次doorbell
746 29 : if (isLastWr || wrNum == wqePerDoorBell_) {
747 29 : u32 dbIndex = static_cast<u32>(opRsp.db.dbIndex);
748 29 : u64 dbInfo = static_cast<u64>(opRsp.db.dbInfo);
749 29 : HCCL_DEBUG(
750 : "[SendRecvExecutor][MultiWqeOneDoorBellSend] Start RDMADBSend, dbIndex[%u], dbInfo[%llu], wrNum[%u], "
751 : "isLastMR[%d].",
752 : dbIndex, dbInfo, wrNum, isLastWr);
753 29 : ret = hrtRDMADBSend(dbIndex, dbInfo, stream_);
754 29 : CHK_PRT_RET(
755 : ret != HCCL_SUCCESS,
756 : HCCL_ERROR(
757 : "[SendRecvExecutor][MultiWqeOneDoorBellSend]errNo[0x%016llx] In lbv exp op base mode, "
758 : "rdma send failed. dbIndex[%u] dbInfo[%llu]",
759 : HCCL_ERROR_CODE(ret), dbIndex, dbInfo),
760 : ret);
761 29 : wrNum = 0;
762 : }
763 29 : return HCCL_SUCCESS;
764 : }
765 :
766 19 : HcclResult SendRecvExecutor::WaitPutMR()
767 : {
768 19 : HcclResult ret = HCCL_SUCCESS;
769 :
770 19 : ret = WaitSignalUnlimitedTime(doneNotify_);
771 19 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][WaitPutMR] Wait done failed"), ret);
772 :
773 38 : ret = RecordNotify(
774 19 : remoteSyncMemAck_.addr, remoteSyncMemAck_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
775 19 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][WaitPutMR] Record ack failed"), ret);
776 :
777 19 : return HCCL_SUCCESS;
778 : }
779 :
780 0 : HcclResult SendRecvExecutor::ProcessRCQ(AscendMrInfo* lastMRInfo)
781 : {
782 0 : HcclResult ret = HCCL_SUCCESS;
783 0 : u64 remainingSize = lastMRInfo->size;
784 0 : u64 localAddr = lastMRInfo->addr;
785 0 : u64 offSet = 0;
786 0 : while (remainingSize > MAX_RDMA_WQE_SIZE) {
787 0 : remainingSize -= MAX_RDMA_WQE_SIZE;
788 0 : offSet += MAX_RDMA_WQE_SIZE;
789 : }
790 0 : std::vector<struct RecvWrlistData> recvWrVec(1);
791 0 : recvWrVec[0].wrId = localAddr;
792 0 : recvWrVec[0].memList.addr = localAddr + offSet;
793 0 : recvWrVec[0].memList.len = remainingSize;
794 0 : recvWrVec[0].memList.lkey = lastMRInfo->key;
795 :
796 0 : struct RecvWrlistData* recvWr = recvWrVec.data();
797 0 : u32 completeNum = 0;
798 0 : ret = hrtRaRecvWrlist(qpHandle_, recvWr, 1, &completeNum);
799 0 : if (ret == HCCL_SUCCESS && completeNum == 1) {
800 0 : HCCL_INFO("[SendRecvExecutor][TestBatchPutMR] Exec hrtRaRecvWrlist success.");
801 : } else {
802 0 : HCCL_ERROR(
803 : "[SendRecvExecutor][TestBatchPutMR] In RdmaDataTransport, hrtRaRecvWrlist failed. ret[%d], "
804 : "completeNum[%d].",
805 : ret, completeNum);
806 0 : return HCCL_E_NETWORK;
807 : }
808 :
809 0 : CHK_RET(PollCq());
810 0 : return HCCL_SUCCESS;
811 0 : }
812 :
813 19 : HcclResult SendRecvExecutor::WaitSignalUnlimitedTime(HcclRtSignal signal)
814 : {
815 : // 超时时间配置成0，代表永不超时
816 19 : CHK_RET(hrtNotifyWaitWithTimeOut(static_cast<HcclRtNotify>(signal), stream_, 0));
817 19 : return HCCL_SUCCESS;
818 : }
819 :
820 0 : HcclResult SendRecvExecutor::WaitPutMROnlyWait()
821 : {
822 0 : HcclResult ret = HCCL_SUCCESS;
823 :
824 0 : ret = WaitSignalUnlimitedTime(doneNotify_);
825 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][WaitPutMROnlyWait] Wait done failed"), ret);
826 :
827 0 : return HCCL_SUCCESS;
828 : }
829 :
830 0 : HcclResult SendRecvExecutor::WaitPutMROnlyRecord()
831 : {
832 0 : HcclResult ret = HCCL_SUCCESS;
833 :
834 0 : ret = RecordNotify(
835 0 : remoteSyncMemAck_.addr, remoteSyncMemAck_.lkey, notifySrcMem_.addr, notifySrcMem_.lkey, notifySize_);
836 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[SendRecvExecutor][WaitPutMROnlyWait] Record ack failed"), ret);
837 :
838 0 : return HCCL_SUCCESS;
839 : }
840 :
841 10 : HcclResult SendRecvExecutor::OneSideBatchPutMR(u32 num, AscendMrInfo* putMRList, AscendMrInfo* remoteMRList)
842 : {
843 10 : CHK_RET(hrtGetNotifySize(notifySize_));
844 10 : if (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET) {
845 0 : notifyWaitMode_ = SyncMode::CONFIGURABLE_TIMEWAITSYNCMODE;
846 : }
847 10 : CHK_RET(TypicalSyncMem::GetInstance().GetNotifyHandle(reinterpret_cast<u64>(localSyncMemAck_.addr), ackNotify_));
848 10 : CHK_PTR_NULL(ackNotify_);
849 10 : HCCL_INFO(
850 : "[OneSideBatchPutMR] notifySize[%u], notifyWaitMode[%d], "
851 : "ackNotify[%p], remoteNotifyValueMem addr[%p], remoteNotifyValueMem len[%llu], remoteNotifyValueMem key[%u], "
852 : "wqePerDoorBell[%u]",
853 : notifySize_, notifyWaitMode_, ackNotify_, remoteNotifyValueMem_.addr, remoteNotifyValueMem_.size,
854 : remoteNotifyValueMem_.lkey, wqePerDoorBell_);
855 :
856 10 : u32 sendWrNum = 0;
857 20 : for (u32 i = 0; i < num; i++) {
858 10 : if (i != num - 1) {
859 0 : CHK_RET(PayLoadMR(putMRList + i, remoteMRList + i, sendWrNum));
860 : } else {
861 : // 最后一组数据特殊处理，发送立即数
862 10 : CHK_RET(PayLoadMR(putMRList + i, remoteMRList + i, sendWrNum, true));
863 : }
864 : }
865 :
866 20 : HcclResult ret = RecordNotify(
867 10 : remoteNotifyValueMem_.addr, remoteNotifyValueMem_.lkey, localSyncMemAck_.addr, localSyncMemAck_.lkey,
868 10 : notifySize_, RA_WR_RDMA_READ, RA_SEND_SIGNALED | RA_SEND_FENCE);
869 10 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OneSideBatchPutMR] Record ack failed"), ret);
870 :
871 10 : ret = WaitSignal(ackNotify_);
872 10 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[OneSideBatchPutMR] Wait ack failed"), ret);
873 :
874 10 : return ret;
875 : }
876 :
877 : } // namespace hccl
|
