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 "aicpu_dispatcher.h"
12 :
13 : #include "common/aicpu_sqe_context.h"
14 : #include "common/aicpu_hccl_common.h"
15 : #include "adapter_hal_pub.h"
16 :
17 : using namespace hccl;
18 :
19 3599 : HcclResult AicpuDispatcher::SignalWait(u16 streamId, u16 notifyId, bool innerChip, bool preNotify)
20 : {
21 3599 : auto ctx = AicpuGetComContext();
22 3599 : AicpuComSignalInfo* notifyInfo
23 5168 : = innerChip ? (preNotify ? &ctx->noIpcPreNotify[notifyId] : &ctx->noIpcPostNotify[notifyId]) :
24 1569 : (preNotify ? &ctx->ipcPreWaitNotify[notifyId] : &ctx->ipcPostWaitNotify[notifyId]);
25 3599 : return SignalWaitWithNotify(streamId, notifyId, innerChip, notifyInfo);
26 : }
27 :
28 18 : HcclResult AicpuDispatcher::AicpuUnfoldSignalWait(u16 streamId, u16 notifyId, bool innerChip)
29 : {
30 18 : auto ctx = AicpuGetComContext();
31 18 : AicpuComSignalInfo* notifyInfo = &ctx->aicpuOpNotify[notifyId];
32 18 : return SignalWaitWithNotify(streamId, notifyId, innerChip, notifyInfo);
33 : }
34 :
35 : HcclResult
36 3617 : AicpuDispatcher::SignalWaitWithNotify(u16 streamId, u16 notifyId, bool innerChip, AicpuComSignalInfo* notifyInfo)
37 : {
38 3617 : auto ctx = AicpuGetComContext();
39 3617 : HcclComStreamInfo* streamInfo = &ctx->streamInfo[streamId];
40 3617 : uint8_t* sqeBuffer = nullptr;
41 3617 : uint8_t* sqeTypeAddr = nullptr;
42 3617 : uint16_t taskId = 0U;
43 3617 : CHK_RET(AicpuSqeContext::GetNextSqeBufferAddr(streamId, sqeBuffer, sqeTypeAddr, taskId));
44 3617 : if (innerChip || (ctx->devType != DevType::DEV_TYPE_310P1 && ctx->devType != DevType::DEV_TYPE_310P3)) {
45 3614 : AicpuAddOneNotifyWaitSqe addOneNotifyWaitSqe = AicpuGetAddOneNotifyWaitSqe();
46 3614 : if (addOneNotifyWaitSqe == nullptr) {
47 0 : HCCL_ERROR("AicpuAddOneNotifyWaitSqe is null.");
48 0 : return HCCL_SUCCESS;
49 : }
50 3614 : if (ctx->debugMode == MC2_DEBUG_NOTIFY_WAIT_TIMEOUT) {
51 0 : addOneNotifyWaitSqe(
52 0 : streamInfo->actualStreamId, taskId, INVALID_U64, sqeBuffer, sqeTypeAddr,
53 0 : ctx->dfxExtendInfo.dfxTimeOutConfig);
54 : } else {
55 3614 : addOneNotifyWaitSqe(
56 3614 : streamInfo->actualStreamId, taskId, notifyInfo->actualNotifyId, sqeBuffer, sqeTypeAddr,
57 3614 : ctx->dfxExtendInfo.dfxTimeOutConfig);
58 : }
59 :
60 3614 : if (innerChip) {
61 2031 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, ctx->rankId));
62 : } else {
63 1583 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, notifyId));
64 : }
65 3614 : } else {
66 3 : u32 notifyRevisedOffset = 15U; // eventid偏移15位后为1
67 3 : u32 notifyGetEventId = 0x3FFU; // 取低15位
68 3 : if ((static_cast<u32>(notifyInfo->actualNotifyId) >> notifyRevisedOffset) != 0) {
69 2 : AicpuAddOneEventWaitSqe addOneEventWaitSqe = AicpuGetAddOneEventWaitSqe();
70 2 : if (addOneEventWaitSqe == nullptr) {
71 0 : HCCL_ERROR("addOneEventWaitSqe is null");
72 0 : return HCCL_SUCCESS;
73 : }
74 2 : addOneEventWaitSqe(
75 2 : streamInfo->actualStreamId, (static_cast<u32>(notifyInfo->actualNotifyId) & notifyGetEventId), taskId,
76 : sqeBuffer, sqeTypeAddr);
77 2 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, notifyId));
78 :
79 2 : uint8_t* sqeBuffer1 = nullptr;
80 2 : uint8_t* sqeTypeAddr1 = nullptr;
81 2 : CHK_RET(AicpuSqeContext::GetNextSqeBufferAddr(streamId, sqeBuffer1, sqeTypeAddr1, taskId));
82 :
83 2 : AicpuAddOneEventResetSqe addOneEventResetSqe = AicpuGetAddOneEventResetSqe();
84 2 : if (addOneEventResetSqe == nullptr) {
85 0 : HCCL_ERROR("addOneEventResetSqe is null");
86 0 : return HCCL_SUCCESS;
87 : }
88 2 : addOneEventResetSqe(
89 2 : streamInfo->actualStreamId, (static_cast<u32>(notifyInfo->actualNotifyId) & notifyGetEventId), taskId,
90 : streamId, 0, notifyInfo->address, sqeBuffer1, sqeTypeAddr1);
91 2 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, notifyId));
92 : } else {
93 1 : HCCL_WARNING("SignalWait id is not event, please check %d", notifyInfo->actualNotifyId);
94 : }
95 : }
96 :
97 3617 : return HCCL_SUCCESS;
98 : }
99 :
100 3599 : HcclResult AicpuDispatcher::SignalRecord(u16 streamId, u16 notifyId, bool innerChip, bool preNotify)
101 : {
102 3599 : auto ctx = AicpuGetComContext();
103 3599 : AicpuComSignalInfo* notifyInfo
104 5168 : = innerChip ? (preNotify ? &ctx->noIpcPreNotify[notifyId] : &ctx->noIpcPostNotify[notifyId]) :
105 1569 : (preNotify ? &ctx->ipcPreRecordNotify[notifyId] : &ctx->ipcPostRecordNotify[notifyId]);
106 3599 : return SignalRecordWithNotify(streamId, notifyId, innerChip, notifyInfo);
107 : }
108 :
109 15 : HcclResult AicpuDispatcher::AicpuUnfoldSignalRecord(u16 streamId, u16 notifyId, bool innerChip)
110 : {
111 15 : auto ctx = AicpuGetComContext();
112 15 : AicpuComSignalInfo* notifyInfo = &ctx->aicpuOpNotify[notifyId];
113 15 : return SignalRecordWithNotify(streamId, notifyId, innerChip, notifyInfo);
114 : }
115 :
116 : HcclResult
117 3614 : AicpuDispatcher::SignalRecordWithNotify(u16 streamId, u16 notifyId, bool innerChip, AicpuComSignalInfo* notifyInfo)
118 : {
119 3614 : auto ctx = AicpuGetComContext();
120 3614 : HcclComStreamInfo* streamInfo = &ctx->streamInfo[streamId];
121 3614 : uint8_t* sqeBuffer = nullptr;
122 3614 : uint8_t* sqeTypeAddr = nullptr;
123 3614 : uint16_t taskId = 0U;
124 3614 : CHK_RET(AicpuSqeContext::GetNextSqeBufferAddr(streamId, sqeBuffer, sqeTypeAddr, taskId));
125 :
126 3614 : if (innerChip) {
127 2030 : AicpuAddOneRecordSqe addOneRecordSqe = AicpuGetAddOneRecordSqe();
128 2030 : if (addOneRecordSqe == nullptr) {
129 0 : HCCL_ERROR("AicpuAddOneRecordSqe is null");
130 0 : return HCCL_SUCCESS;
131 : }
132 2030 : addOneRecordSqe(streamInfo->actualStreamId, taskId, notifyInfo->actualNotifyId, sqeBuffer, sqeTypeAddr);
133 2030 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, ctx->rankId));
134 : } else {
135 1584 : AicpuAddOneWriteValueRecordSqe addOneWriteValueRecordSqe = AicpuGetAddOneWriteValueRecordSqe();
136 1584 : if (addOneWriteValueRecordSqe == nullptr) {
137 0 : HCCL_ERROR("AicpuAddOneWriteValueRecordSqe is null");
138 0 : return HCCL_SUCCESS;
139 : }
140 1584 : addOneWriteValueRecordSqe(streamInfo->actualStreamId, taskId, notifyInfo->address, sqeBuffer, sqeTypeAddr);
141 1584 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, notifyId));
142 : }
143 3614 : return HCCL_SUCCESS;
144 : }
145 :
146 942 : HcclResult AicpuDispatcher::CopyData(
147 : u16 streamId, void* src, void* dst, u32 len, HcclDataType dataType, HcclReduceOp reduceOp, u32 remoteRank)
148 : {
149 942 : if (len == 0) {
150 85 : return HCCL_SUCCESS;
151 : }
152 857 : CHK_PTR_NULL(src);
153 855 : CHK_PTR_NULL(dst);
154 854 : auto ctx = AicpuGetComContext();
155 854 : HcclComStreamInfo* streamInfo = &ctx->streamInfo[streamId];
156 :
157 854 : aclDataType rtDataType = DT_MAP_TABLE[dataType];
158 854 : aclrtReduceKind rtReduceOp = RK_MAP_TABLE[reduceOp];
159 :
160 854 : uint8_t* sqeBuffer = nullptr;
161 854 : uint8_t* sqeTypeAddr = nullptr;
162 854 : uint16_t taskId = 0U;
163 854 : CHK_RET(AicpuSqeContext::GetNextSqeBufferAddr(streamId, sqeBuffer, sqeTypeAddr, taskId));
164 :
165 854 : AicpuAddOneMemcpySqe addOneMemcpySqe = AicpuGetAddOneMemcpySqe();
166 854 : if (addOneMemcpySqe == nullptr) {
167 0 : HCCL_ERROR("addOneMemcpySqe is null");
168 0 : return HCCL_SUCCESS;
169 : }
170 854 : if (ctx->debugMode == MC2_DEBUG_SDMA_ERROR) {
171 0 : src = nullptr;
172 : }
173 854 : addOneMemcpySqe(
174 854 : streamInfo->actualStreamId, taskId, src, len, rtDataType, rtReduceOp, dst, 0, ctx->ssid, ctx->devId,
175 : ctx->overflowAddr, static_cast<uint8_t>(LinkType::LINK_RESERVED), sqeBuffer, sqeTypeAddr, SDMA_QOS_DEFAULT);
176 854 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, (remoteRank << 16) + static_cast<uint32_t>(dataType))); // 16 bit
177 854 : return HCCL_SUCCESS;
178 : }
179 :
180 306 : HcclResult AicpuDispatcher::CopyData(
181 : uint16_t streamId, u64 src, u64 dst, uint32_t len, HcclDataType dataType, HcclReduceOp reduceOp, u32 remoteRank)
182 : {
183 306 : return CopyData(
184 306 : streamId, reinterpret_cast<void*>(src), reinterpret_cast<void*>(dst), len, dataType, reduceOp, remoteRank);
185 : }
186 :
187 713 : HcclResult AicpuDispatcher::LaunchTask(uint32_t streamId)
188 : {
189 713 : auto ctx = AicpuGetComContext();
190 713 : HcclComStreamInfo* streamInfo = &ctx->streamInfo[streamId];
191 713 : auto& sqeContextBuffer = GetSqeContext()->buffPtr[streamId];
192 713 : const auto cnt = sqeContextBuffer.sqeCnt;
193 713 : if (cnt == 0U) {
194 74 : HCCL_DEBUG("no sqe, rankid:%u, streamId:%d, sqId:%u", streamId, streamInfo->actualStreamId, streamInfo->sqId);
195 74 : return HCCL_SUCCESS;
196 : }
197 639 : auto& head = sqeContextBuffer.sqHead;
198 639 : auto& tail = sqeContextBuffer.sqTail;
199 639 : u32 newTail = (tail + cnt) % streamInfo->sqDepth;
200 639 : HCCL_INFO("Before send sqe:%d cnt:%u head:%u curtail:%u newTail:%u", streamInfo->sqId, cnt, head, tail, newTail);
201 :
202 639 : u64 startUsec = GetCurCpuTimestamp();
203 639 : while ((tail < head ? streamInfo->sqDepth : 0U) + tail - head + cnt >= streamInfo->sqDepth) { // 存在回绕
204 0 : CHK_RET(QuerySqStatusByType(ctx->devId, streamInfo->sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
205 0 : if (GetCurCpuTimestamp() - startUsec > NSEC_PER_SEC * ctx->dfxExtendInfo.dfxTimeOutConfig.sqFullWaitTimeOut) {
206 0 : HCCL_ERROR(
207 : "Rtsq full, timeout %lus. cur head:%u, sqId:%d", ctx->dfxExtendInfo.dfxTimeOutConfig.sqFullWaitTimeOut,
208 : head, streamInfo->sqId);
209 0 : return HCCL_E_INTERNAL;
210 : }
211 : }
212 :
213 639 : auto memcpyFunc = [&](uint32_t dst, uint32_t dstMax, uint32_t src, uint32_t length) -> HcclResult {
214 639 : HCCL_DEBUG("Memcpy rank:%u , dst:%u, dstMax:%u, src:%u, length:%u", streamId, dst, dstMax, src, length);
215 639 : if (length == 0U) {
216 0 : return HCCL_SUCCESS;
217 : }
218 1278 : errno_t ret = memcpy_s(
219 639 : reinterpret_cast<uint8_t*>(streamInfo->sqBaseAddr) + dst * AC_SQE_SIZE, dstMax * AC_SQE_SIZE,
220 639 : sqeContextBuffer.localBuff + src * AC_SQE_SIZE, length * AC_SQE_SIZE);
221 639 : if (ret != EOK) {
222 0 : HCCL_ERROR("Memcpy ret %d, dst:%u, dstMax:%u, src:%u, length:%u", ret, dst, dstMax, src, length);
223 0 : return HCCL_E_MEMORY;
224 : }
225 639 : return HCCL_SUCCESS;
226 639 : };
227 639 : uint32_t left = streamInfo->sqDepth - tail; // sqeAddr 剩余空间
228 639 : const auto tailSqeIdx = sqeContextBuffer.tailSqeIdx;
229 639 : HCCL_INFO("cpy sqe, left:%u, tailSqeId:%u, cnt:%u", left, tailSqeIdx, cnt);
230 639 : if (cnt <= left) { // 剩余buffer放得下新增sqe
231 639 : CHK_RET(memcpyFunc(tail, left, tailSqeIdx - cnt, cnt));
232 : } else {
233 0 : CHK_RET(memcpyFunc(tail, left, tailSqeIdx - cnt, left));
234 0 : CHK_RET(memcpyFunc(0, streamInfo->sqDepth, tailSqeIdx - cnt + left, cnt - left));
235 : }
236 639 : CHK_RET(ConfigSqStatusByType(ctx->devId, streamInfo->sqId, DRV_SQCQ_PROP_SQ_TAIL, newTail));
237 :
238 639 : tail = newTail;
239 639 : HCCL_INFO(
240 : "After send sqe:%d, sqe_num:%u, curHead:%u, curtail:%u, sqeCnt:%u, tailSqeIdx:%u", streamInfo->sqId, cnt, head,
241 : tail, sqeContextBuffer.sqeCnt, sqeContextBuffer.tailSqeIdx);
242 639 : sqeContextBuffer.sqeCnt = 0;
243 : // StartMC2MaintenanceThread函数如果为空，说明是老的驱动包，为了解决老的驱动包可能的异常cq占满物理cq队列的情况，
244 : // 我们这里使用物理cq查询接口来清理队列； 如果是新的驱动包，会在StartMC2MaintenanceThread线程中使用logic
245 : // cq进行查询并解析
246 639 : if (!IsSupportStartMC2MaintenanceThread()
247 639 : && (ctx->devType != DevType::DEV_TYPE_310P1 && ctx->devType != DevType::DEV_TYPE_310P3)) {
248 : CqeQueryInput cqeQueryInput;
249 0 : cqeQueryInput.devId = ctx->devId;
250 0 : cqeQueryInput.streamId = streamInfo->actualStreamId;
251 0 : cqeQueryInput.sqId = streamInfo->sqId;
252 0 : cqeQueryInput.cqId = streamInfo->sqId; // 使用sqid替代cqid，只有在sq cq成对申请，sqid cqid一样时才可以
253 0 : cqeQueryInput.type = static_cast<uint32_t>(DRV_NORMAL_TYPE);
254 : uint8_t tmpAddr[MAX_REPORT_CNT * 16]; // 16 cqe byte size
255 0 : cqeQueryInput.cqeAddr = tmpAddr;
256 0 : HCCL_DEBUG("Start to call cq report with [%s]", cqeQueryInput.ToString().c_str());
257 : rtLogicCqReport_t cqeException;
258 0 : (void)CqReportRecv(cqeQueryInput, cqeException);
259 : }
260 639 : return HCCL_SUCCESS;
261 : }
262 :
263 168 : HcclResult AicpuDispatcher::AddCcoreWait(uint16_t streamId, u64 waitAddr, uint32_t turnNum, bool isLast)
264 : {
265 168 : auto ctx = AicpuGetComContext();
266 168 : HcclComStreamInfo* streamInfo = &ctx->streamInfo[streamId];
267 :
268 168 : uint8_t* sqeBuffer = nullptr;
269 168 : uint8_t* sqeTypeAddr = nullptr;
270 168 : uint16_t taskId = 0U;
271 168 : CHK_RET(AicpuSqeContext::GetNextSqeBufferAddr(streamId, sqeBuffer, sqeTypeAddr, taskId));
272 :
273 168 : HCCL_INFO(
274 : "[SQE]Add ccore wait addr %p, workSpaceAddr %p, notifyOff %u, turnNum %u, streamId=%u, isLast=%d", waitAddr,
275 : ctx->workSpaceAddr, ctx->notifyOff, turnNum, streamInfo->actualStreamId, isLast);
276 168 : if (ctx->debugMode == MC2_DEBUG_COMMIT_TIMEOUT) {
277 0 : ctx->turnValue[turnNum] = 0xFF;
278 : }
279 168 : AddOneWaitStartSqe(
280 168 : streamInfo->actualStreamId, taskId, waitAddr, reinterpret_cast<u64>(&ctx->turnValue[turnNum]), isLast,
281 : reinterpret_cast<rtStarsCcoreWaitStartSqe_t*>(sqeBuffer), sqeTypeAddr);
282 168 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, (turnNum << 16) + static_cast<uint32_t>(isLast))); // 16 bit
283 168 : return HCCL_SUCCESS;
284 : }
285 :
286 82 : HcclResult AicpuDispatcher::AddWaitStartTaskOnMainStream(u16 streamId)
287 : {
288 82 : auto ctx = AicpuGetComContext();
289 : // 保持和AIC的间消息长度一致，每隔64字节(sizeof(u8)*AC_SQE_SIZE)写一个地址。
290 82 : u64 waitAddr = 0;
291 82 : uint32_t turnNum = 0;
292 82 : bool isLast = 0;
293 82 : if (ctx->preparePosition == TASK_PREPARE_KERNEL) {
294 24 : waitAddr = ctx->workSpaceAddr + offsetof(HcclApi::HcclMsgArea, commMsg.singleMsg.commitTurnCnt)
295 24 : + ctx->msgPosForKernel * sizeof(HcclApi::TurnCnt) + offsetof(HcclApi::TurnCnt, cnt);
296 :
297 24 : turnNum = ctx->curTurnCntForKernel;
298 24 : isLast = ctx->curTurnCntForKernel >= ctx->totalTurnCntForKernel;
299 24 : HCCL_INFO(
300 : "aicpu kernel mode, curTurnCnt %u, totalTurnCnt %u", ctx->curTurnCntForKernel, ctx->totalTurnCntForKernel);
301 : } else {
302 58 : waitAddr = ctx->workSpaceAddr + ctx->notifyOff + offsetof(AivAicpuOpParam, sendCnt);
303 58 : turnNum = (ctx->curTurnCnt + 1);
304 58 : isLast = (ctx->curTurnCnt + 1 >= ctx->totalTurnCnt);
305 : }
306 82 : return AddCcoreWait(streamId, waitAddr, turnNum, isLast);
307 : }
308 :
309 167 : HcclResult AicpuDispatcher::AddCcoreNotify(uint16_t streamId, uint32_t turnNum)
310 : {
311 167 : auto ctx = AicpuGetComContext();
312 167 : HcclComStreamInfo* streamInfo = &ctx->streamInfo[streamId];
313 167 : uint8_t* sqeBuffer = nullptr;
314 167 : uint8_t* sqeTypeAddr = nullptr;
315 167 : uint16_t taskId = 0U;
316 167 : CHK_RET(AicpuSqeContext::GetNextSqeBufferAddr(streamId, sqeBuffer, sqeTypeAddr, taskId));
317 :
318 167 : u64 recordAddr = 0;
319 167 : if (ctx->preparePosition == TASK_PREPARE_KERNEL) {
320 23 : recordAddr = ctx->workSpaceAddr + offsetof(HcclApi::HcclMsgArea, commMsg.singleMsg.finishedTurnCnt)
321 23 : + ctx->msgPosForKernel * sizeof(HcclApi::TurnCnt) + offsetof(HcclApi::TurnCnt, cnt);
322 : } else {
323 144 : recordAddr = ctx->workSpaceAddr + ctx->notifyOff + ctx->notifyBeginCnt * AC_SQE_SIZE
324 : + offsetof(AivAicpuOpParam, rcvCnt);
325 : }
326 167 : HCCL_INFO(
327 : "[SQE]Add ccore notify recordAddr %p, workSpaceAddr %p, notifyOff %u, notifyBeginCnt %u,"
328 : "streamId=%u, curTurnCnt %u, turnNum %u, preparePosition %u, msgPos %u",
329 : recordAddr, ctx->workSpaceAddr, ctx->notifyOff, ctx->notifyBeginCnt, streamInfo->actualStreamId,
330 : ctx->curTurnCnt, turnNum, ctx->preparePosition, ctx->msgPosForKernel);
331 :
332 167 : if (ctx->debugMode == MC2_DEBUG_AICORE_WAIT_TIMEOUT) {
333 0 : ctx->turnValue[turnNum] = 0;
334 : }
335 167 : AddOneWriteValueStartSqe(
336 167 : streamInfo->actualStreamId, taskId, recordAddr, reinterpret_cast<u64>(&ctx->turnValue[turnNum]),
337 : reinterpret_cast<rtStarsCcoreWriteValueSqe_t*>(sqeBuffer), sqeTypeAddr);
338 167 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, turnNum));
339 167 : return HCCL_SUCCESS;
340 : }
341 :
342 81 : HcclResult AicpuDispatcher::AddExecEndTaskOnMainStream(u16 streamId)
343 : {
344 81 : auto ctx = AicpuGetComContext();
345 81 : uint32_t turnNum = ctx->preparePosition == TASK_PREPARE_KERNEL ? ctx->curTurnCntForKernel : ctx->curTurnCnt;
346 81 : return AddCcoreNotify(streamId, turnNum);
347 : }
348 :
349 49 : HcclResult AicpuDispatcher::AddAllEndTaskOnMainStream(u16 streamId)
350 : {
351 49 : auto ctx = AicpuGetComContext();
352 49 : HcclComStreamInfo* streamInfo = &ctx->streamInfo[streamId];
353 49 : uint8_t* sqeBuffer = nullptr;
354 49 : uint8_t* sqeTypeAddr = nullptr;
355 49 : uint16_t taskId = 0U;
356 49 : CHK_RET(AicpuSqeContext::GetNextSqeBufferAddr(streamId, sqeBuffer, sqeTypeAddr, taskId));
357 :
358 49 : AicpuAddOneRecordSqe addOneRecordSqe = AicpuGetAddOneRecordSqe();
359 49 : if (addOneRecordSqe == nullptr) {
360 0 : HCCL_ERROR("AicpuAddOneRecordSqe is null");
361 0 : return HCCL_SUCCESS;
362 : }
363 49 : HCCL_INFO("[SQE]Add all end task kfcNotifyId %lu, streamId %d", ctx->kfcNotifyId, streamInfo->actualStreamId);
364 49 : addOneRecordSqe(streamInfo->actualStreamId, taskId, ctx->kfcNotifyId, sqeBuffer, sqeTypeAddr);
365 49 : CHK_RET(AicpuSqeContext::RecordAddInfo(streamId, ctx->rankId));
366 49 : return HCCL_SUCCESS;
367 : }
368 :
369 1 : HcclResult AicpuDispatcher::RdmaSend(uint16_t streamId, u64 dbInfo, u64 dbAddr, u32 userRank)
370 : {
371 1 : auto ctx = AicpuGetComContext();
372 1 : HcclComStreamInfo* streamInfo = &ctx->streamInfo[streamId];
373 1 : uint8_t* sqeBuffer = nullptr;
374 1 : uint8_t* sqeTypeAddr = nullptr;
375 1 : uint16_t taskId = 0U;
376 1 : CHK_RET(AicpuSqeContext::GetNextSqeBufferAddr(streamId, sqeBuffer, sqeTypeAddr, taskId));
377 :
378 1 : AicpuAddOneRdmaDbSendSqe AddOneRdmaDbSendSqe = AicpuGetAddOneRdmaDbSendSqe();
379 1 : if (AddOneRdmaDbSendSqe == nullptr) {
380 0 : HCCL_ERROR("[AicpuDispatcher][RdmaSend] AddOneRdmaDbSendSqe is null");
381 0 : return HCCL_E_PTR;
382 : }
383 1 : AddOneRdmaDbSendSqe(
384 1 : streamInfo->actualStreamId, taskId, dbInfo, dbAddr, 0, static_cast<uint8_t>(hccl::RdmaType::RDMA_TYPE_RESERVED),
385 : sqeBuffer, sqeTypeAddr);
386 :
387 1 : HCCL_INFO(
388 : "[AicpuDispatcher][RdmaSend] Call RdmaSend. para: rankId[%u] "
389 : "taskId[%u], streamId[%u]",
390 : userRank, taskId, streamInfo->actualStreamId);
391 :
392 1 : return HCCL_SUCCESS;
393 : }
|
