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