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