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 <chrono>
12 : #include <unordered_map>
13 : #include "rtsq_a5.h"
14 : #include "log.h"
15 : #include "exception_util.h"
16 : #include "internal_exception.h"
17 : #include "sqe_build_a5.h"
18 : #include "sqe.h"
19 : #ifdef CCL_KERNEL_AICPU
20 : #include "aicpu_ts_primitives_c_adpt.h"
21 : #endif
22 : #include "aicpu_task_utils.h"
23 :
24 : namespace Hccl {
25 : using namespace std;
26 : constexpr u32 RTSQ_A5_PART_ID = 0;
27 : constexpr u32 PRINT_INTERVAL = 30;
28 :
29 288 : RtsqA5::RtsqA5(u32 devPhyId, u32 streamId, u32 sqId) : RtsqBase(devPhyId, streamId, sqId) { SetTaskIdBySqeId(); }
30 :
31 0 : RtsqA5::RtsqA5(u32 devPhyId, u32 streamId, u32 sqId, bool launchFlag) : RtsqBase(devPhyId, streamId, sqId)
32 : {
33 0 : SetTaskIdBySqeId();
34 0 : launchFlag_ = launchFlag;
35 0 : }
36 :
37 1 : void RtsqA5::Reset()
38 : {
39 1 : RtsqBase::Reset();
40 1 : pendingSqeCnt = 0;
41 1 : s32 sRet = memset_s(locBuf, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT, 0, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT);
42 1 : if (UNLIKELY(sRet != EOK)) {
43 0 : auto msg = StringFormat("[RtsqA5][Reset] locBuf memset fail. errorno[%d]", sRet);
44 0 : THROW<InternalException>(msg);
45 0 : }
46 3 : HCCL_INFO("[NsRecovery]RtsqA5::%s success", __func__);
47 1 : }
48 :
49 : // 计算head和tail之间的距离
50 11 : u32 RtsqA5::GetTailToHeadDist() const
51 : {
52 11 : if (UNLIKELY(sqHead_ == sqTail_)) { // 头尾相同，则距离大小为sq深度
53 8 : return sqDepth_;
54 : }
55 3 : return (sqTail_ < sqHead_) ? (sqHead_ - sqTail_) : (sqDepth_ - (sqTail_ - sqHead_));
56 : }
57 :
58 7 : void RtsqA5::MakeSureAvailableSpace()
59 : {
60 7 : u32 availableSpace = GetTailToHeadDist();
61 7 : auto startTime = std::chrono::steady_clock::now();
62 : #ifdef CCL_KERNEL_AICPU
63 : sqFullTimeout_ = GetSqFullTimeOut();
64 : #endif
65 7 : const std::chrono::seconds printInterval(PRINT_INTERVAL); // 打印间隔30s
66 7 : auto lastPrintTime = std::chrono::steady_clock::now() - printInterval;
67 21 : HCCL_INFO(
68 : "[%s]sqId:%u, sqFullTimeout_: %u s, sqHead:%u, sqTail:%u, pendingSqeCnt:%u", __func__, sqId_, sqFullTimeout_,
69 : sqHead_, sqTail_, pendingSqeCnt);
70 :
71 7 : while (availableSpace <= pendingSqeCnt) {
72 0 : sqHead_ = QuerySqHead();
73 0 : availableSpace = GetTailToHeadDist();
74 0 : if (availableSpace > pendingSqeCnt) {
75 0 : break; // 避免head没更新导致假反压
76 : }
77 :
78 0 : auto curTime = std::chrono::steady_clock::now();
79 0 : if (UNLIKELY(curTime - lastPrintTime >= printInterval)) {
80 0 : HCCL_RUN_INFO(
81 : "[%s]while loop, sqId:%u, sqHead:%u, sqTail:%u, availableSpace:%u, pendingSqeCnt:%u, "
82 : "sqFullTimeout_:%u s",
83 : __func__, sqId_, sqHead_, sqTail_, availableSpace, pendingSqeCnt, sqFullTimeout_);
84 0 : lastPrintTime = curTime;
85 : }
86 :
87 0 : CheckLaunchTaskStatus(startTime, curTime);
88 : #ifdef CCL_KERNEL_AICPU
89 : HcclResult ret = HandleDispatchAllStreams();
90 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
91 : auto msg
92 : = StringFormat("RtsqA5::%s HandleDispatchAllStreams failed, ret = %d, sqId:%u, ", __func__, ret, sqId_);
93 : HCCL_ERROR("%s", msg.c_str());
94 : THROW<InternalException>(msg);
95 : }
96 : #endif
97 0 : if (checkOpExecStatusCallback_ != nullptr) {
98 0 : checkOpExecStatusCallback_();
99 : }
100 : }
101 7 : }
102 :
103 5 : void RtsqA5::CheckLaunchTaskStatus(
104 : const std::chrono::steady_clock::time_point& startTime, const std::chrono::steady_clock::time_point& curTime)
105 : {
106 5 : bool isTimeout = (sqFullTimeout_ == 0) ? false : ((curTime - startTime) >= std::chrono::seconds(sqFullTimeout_));
107 : // step1 检测是否launch超时，如果超时打印rtsq full的ERROR日志
108 5 : if (UNLIKELY(isTimeout)) {
109 6 : HCCL_ERROR(
110 : "Rtsq full, sqFullTimeout_:[%u s]. sqId:[%u], sqHead:[%u], sqTail:[%u], pendingSqeCnt:[%u]", sqFullTimeout_,
111 : sqId_, sqHead_, sqTail_, pendingSqeCnt);
112 : }
113 :
114 5 : HcclResult checkRet = (checkExecStatusCallback_ != nullptr) ? checkExecStatusCallback_(isTimeout) : HCCL_SUCCESS;
115 : // step2 通信域状态为HCCL_COMM_STATUS_SUSPENDING状态，则终止launch不抛异
116 5 : if (UNLIKELY(checkRet == HCCL_E_SUSPENDING)) {
117 3 : pendingSqeCnt = 0;
118 3 : return;
119 : }
120 : // step3 调用回调检查执行状态：1、如果超时，打印taskException；2、如果通信域不可用，终止launch
121 2 : if (UNLIKELY(isTimeout || checkRet != HCCL_SUCCESS)) {
122 2 : THROW<InternalException>(
123 4 : StringFormat("[%s]stop launch Task, isTimeout[%d], checkRet[%d]", __func__, isTimeout, checkRet));
124 : }
125 : }
126 :
127 11 : void RtsqA5::CopySqeBufToSq(u8* sqeBuf)
128 : {
129 11 : u8* sqCurrAddr = reinterpret_cast<u8*>(sqBaseAddr_) + sqTail_ * RTSQ_SQE_SIZE;
130 11 : if (sqTail_ >= sqHead_) {
131 10 : u32 depthLeft = sqDepth_ - sqTail_;
132 10 : if (pendingSqeCnt <= depthLeft) { // 没有回绕
133 21 : HCCL_INFO(
134 : "RtsqA5::%s copy sqe from sqe buffer, sqId_: %u, streamId_: %u, cur head: %u, cur tail: %u, size: %u, "
135 : "depth remain: %u",
136 : __func__, sqId_, streamId_, sqHead_, sqTail_, pendingSqeCnt, depthLeft);
137 7 : int ret = memcpy_sp(sqCurrAddr, pendingSqeCnt * AC_SQE_SIZE, sqeBuf, pendingSqeCnt * RTSQ_SQE_SIZE);
138 7 : if (UNLIKELY(ret != 0)) {
139 4 : THROW<InternalException>(StringFormat("RtsqA5::%s sqe memcpy_sp failed, ret = %d", __func__, ret));
140 : }
141 : } else {
142 9 : HCCL_INFO(
143 : "RtsqA5::%s copy sqe twice, sqId_: %u, streamId_: %u, cur head: %u, cur tail: %u, cnt: %u, depth "
144 : "remain: %u",
145 : __func__, sqId_, streamId_, sqHead_, sqTail_, pendingSqeCnt, depthLeft);
146 : // 先拷贝rtsq里剩余空间大小
147 3 : int ret = memcpy_sp(sqCurrAddr, depthLeft * AC_SQE_SIZE, sqeBuf, depthLeft * RTSQ_SQE_SIZE);
148 3 : if (ret != 0) {
149 1 : THROW<InternalException>(
150 3 : StringFormat("RtsqA5::%s rtsq remaining space memcpy_sp failed, ret = %d", __func__, ret));
151 : }
152 : // 拷贝剩余sqe
153 2 : ret = memcpy_sp(
154 : reinterpret_cast<u8*>(sqBaseAddr_), sqHead_ * RTSQ_SQE_SIZE, sqeBuf + depthLeft * RTSQ_SQE_SIZE,
155 : (pendingSqeCnt - depthLeft) * AC_SQE_SIZE);
156 2 : if (UNLIKELY(ret != 0)) {
157 0 : THROW<InternalException>(
158 0 : StringFormat("RtsqA5::%s remaining sqe memcpy_sp failed, ret = %d", __func__, ret));
159 : }
160 : }
161 : } else {
162 3 : HCCL_INFO(
163 : "RtsqA5::%s copy sqe from sqe buffer, tail < head, sqId_: %u, streamId_: %u, cur head: %u, cur tail: %u, "
164 : "size: %u",
165 : __func__, sqId_, streamId_, sqHead_, sqTail_, pendingSqeCnt);
166 1 : int ret = memcpy_sp(sqCurrAddr, pendingSqeCnt * AC_SQE_SIZE, sqeBuf, pendingSqeCnt * RTSQ_SQE_SIZE);
167 1 : if (UNLIKELY(ret != 0)) {
168 0 : THROW<InternalException>(StringFormat("RtsqA5::%s sqe memcpy_sp failed, ret = %d", __func__, ret));
169 : }
170 : }
171 8 : }
172 :
173 4 : void RtsqA5::PreLaunchSqeForCache(bool& needCacheTask)
174 : {
175 : // 校验needCacheTaskCallback_
176 : // 注意: A5新流程下needCacheTaskCallback_一定非空; 但A5老流程下不支持aicpu task cache, needCacheTaskCallback_为空;
177 : // 为避免A5老流程报错, 这里为空时跳过执行而非报错
178 4 : needCacheTask = false;
179 4 : if (UNLIKELY(needCacheTaskCallback_ == nullptr)) {
180 12 : HCCL_WARNING("[RtsqA5][PreLaunchSqeForCache] needCacheTaskCallback_ is null, keep needCacheTask as false");
181 : } else {
182 0 : needCacheTask = needCacheTaskCallback_();
183 : }
184 4 : }
185 :
186 0 : void RtsqA5::PostLaunchSqeForCache()
187 : {
188 : // 注意: 只有needCacheTask为true时才调用PostLaunchSqeForCache, 此时一定是A5新流程, 因此addSqeArrayCallback_一定非空
189 0 : if (UNLIKELY(aicpuTsThreadPtr_ == nullptr)) {
190 0 : THROW<InternalException>("[RtsqA5][PostLaunchSqeForCache] aicpuTsThreadPtr_ is null");
191 : }
192 0 : if (UNLIKELY(addSqeArrayCallback_ == nullptr)) {
193 0 : THROW<InternalException>("[RtsqA5][PostLaunchSqeForCache] addSqeArrayCallback_ is null");
194 : }
195 0 : HcclResult ret = addSqeArrayCallback_(this, aicpuTsThreadPtr_, pendingSqeCnt, locBuf, streamId_);
196 0 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
197 0 : THROW<InternalException>("[RtsqA5][PostLaunchSqeForCache] addSqeArrayCallback_ failed, ret %d", ret);
198 : }
199 0 : }
200 :
201 : // 向芯片RTSQ VA中写入 SQE，并触发芯片执行
202 5 : void RtsqA5::LaunchTask()
203 : {
204 15 : HCCL_INFO("RtsqA5::%s: START, pendingSqeCnt[%u]", __func__, pendingSqeCnt);
205 5 : if (pendingSqeCnt == 0) { // 没有SQE ，直接返回
206 6 : HCCL_INFO("RtsqA5::%s: pendingSqeCnt is %u, return", __func__, pendingSqeCnt);
207 2 : return;
208 : }
209 : // 确保 rtsq 有足够空间放pending SQE
210 3 : MakeSureAvailableSpace();
211 :
212 3 : if (pendingSqeCnt == 0) {
213 0 : return;
214 : }
215 :
216 3 : bool needCacheTask = false;
217 3 : PreLaunchSqeForCache(needCacheTask);
218 : // localBuffer拷贝到 RTSQ
219 3 : CopySqeBufToSq(locBuf);
220 :
221 : // 正常展开按需打印SQE
222 3 : if ((UNLIKELY(GetPlfDebugConfigValue() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
223 7 : PLF_CONFIG_DEBUG(
224 : PLF_TASK, "[RtsqA5][LaunchTask] dump %llu generated SQEs in stream[%u]", pendingSqeCnt, streamId_);
225 :
226 3 : int ret = HCCL_SUCCESS;
227 3 : uint8_t* sqePtr = locBuf;
228 9 : for (size_t sqeIdx = 0; sqeIdx < pendingSqeCnt; sqeIdx++) {
229 12 : PLF_CONFIG_DEBUG(PLF_TASK, "[RtsqA5][LaunchTask] %uth generated SQE in stream[%u]", sqeIdx, streamId_);
230 6 : ret = hcomm::AicpuTaskUtils::DumpSqeContent(sqePtr);
231 6 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
232 0 : THROW<InternalException>(StringFormat("RtsqA5::%s DumpSqeContent failed, ret = %d", __func__, ret));
233 : }
234 :
235 6 : sqePtr += RTSQ_SQE_SIZE;
236 : }
237 : }
238 :
239 : // 更新tail，触发芯片执行
240 3 : u32 newTail = (sqTail_ + pendingSqeCnt) % sqDepth_;
241 3 : ConfigSqTail(newTail);
242 3 : sqTail_ = newTail;
243 :
244 : // 缓存sqe
245 3 : if (needCacheTask) {
246 0 : PostLaunchSqeForCache();
247 : }
248 : // 清空本地的locBuffer和sqeCnt数目
249 9 : HCCL_INFO(
250 : "RtsqA5::%s: END, pendingSqeCnt[%u], streamId_[%u] sqHead_[%u] sqTail_[%u]", __func__, pendingSqeCnt, streamId_,
251 : sqHead_, sqTail_);
252 3 : pendingSqeCnt = 0;
253 3 : (void)memset_s(locBuf, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT, 0, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT); // locBuffer清零
254 : }
255 :
256 16 : void RtsqA5::RefreshSqeHeaderTaskField(Rt91095StarsSqeHeader* sqeHeaderPtr)
257 : {
258 16 : SetSqeHeaderTaskFields(sqeHeaderPtr, taskId_);
259 16 : SetTaskIdBySqeId();
260 16 : }
261 :
262 : // 向芯片RTSQ VA中写入aicpu task cache SQE，并触发芯片执行
263 5 : void RtsqA5::LaunchNewTask(uint8_t* sqeArray, uint32_t sqeCount)
264 : {
265 : // 注意: cache命中时才会调用LaunchNewTask, 此时一定不存在pending SQE
266 5 : if (UNLIKELY(pendingSqeCnt > 0)) {
267 2 : THROW<InternalException>(StringFormat(
268 : "RtsqA5::%s: pendingSqeCnt[%u] should be 0 when aicpu task cache hits!", __func__, pendingSqeCnt));
269 : }
270 :
271 : // 临时设置pendingSqeCnt, 用于MakeSureAvailableSpace
272 4 : pendingSqeCnt = sqeCount;
273 :
274 : // 确保 rtsq 有足够空间放pending SQE
275 4 : MakeSureAvailableSpace();
276 :
277 : // sqeArray拷贝到 RTSQ
278 4 : CopySqeBufToSq(sqeArray);
279 :
280 : // 更新tail，触发芯片执行
281 4 : u32 newTail = (sqTail_ + pendingSqeCnt) % sqDepth_;
282 4 : ConfigSqTail(newTail);
283 4 : sqTail_ = newTail;
284 :
285 12 : HCCL_INFO(
286 : "RtsqA5::%s: END, pendingSqeCnt[%u], streamId_[%u] sqHead_[%u] sqTail_[%u]", __func__, pendingSqeCnt, streamId_,
287 : sqHead_, sqTail_);
288 :
289 : // 重置pendingSqeCnt
290 4 : pendingSqeCnt = 0;
291 4 : }
292 :
293 2 : void RtsqA5::TryLaunchTask()
294 : {
295 2 : if (pendingSqeCnt == 0) {
296 1 : return;
297 : }
298 :
299 1 : sqHead_ = QuerySqHead();
300 1 : u32 availableSpace = GetTailToHeadDist();
301 1 : if (availableSpace <= pendingSqeCnt) {
302 0 : return;
303 : }
304 :
305 1 : bool needCacheTask = false;
306 1 : PreLaunchSqeForCache(needCacheTask);
307 :
308 1 : CopySqeBufToSq(locBuf);
309 :
310 1 : u32 newTail = (sqTail_ + pendingSqeCnt) % sqDepth_;
311 1 : ConfigSqTail(newTail);
312 1 : sqTail_ = newTail;
313 :
314 : // 缓存sqe
315 1 : if (needCacheTask) {
316 0 : PostLaunchSqeForCache();
317 : }
318 1 : pendingSqeCnt = 0;
319 1 : (void)memset_s(locBuf, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT, 0, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT);
320 3 : HCCL_INFO(
321 : "RtsqA5::%s: END, pendingSqeCnt[%u], streamId_[%u] sqHead_[%u] sqTail_[%u]", __func__, pendingSqeCnt, streamId_,
322 : sqHead_, sqTail_);
323 : }
324 :
325 36 : u8* RtsqA5::GetCurrSqeBuffer()
326 : {
327 36 : lastSqeAddr_ = sqBaseAddr_ + static_cast<u64>((sqTail_ + pendingSqeCnt) % sqDepth_) * RTSQ_SQE_SIZE;
328 36 : return locBuf + pendingSqeCnt * RTSQ_SQE_SIZE;
329 : }
330 :
331 21 : u64 RtsqA5::GetSqeAddr() const { return lastSqeAddr_; }
332 :
333 33 : void RtsqA5::RefreshInfo()
334 : {
335 33 : SetTaskIdBySqeId();
336 33 : pendingSqeCnt++;
337 :
338 : #ifdef CCL_KERNEL_AICPU
339 : if (launchFlag_ && !IsBatchLaunchMode()) {
340 : LaunchTask();
341 : return;
342 : }
343 : #endif
344 :
345 33 : if (pendingSqeCnt != PER_LAUNCH_SQE_CNT) {
346 33 : return;
347 : }
348 : // 挂起的sqe数量为128个，则需要向芯片RTSQ中写入task
349 0 : LaunchTask();
350 : }
351 :
352 2 : void RtsqA5::NotifyWait(u32 notifyId) { NotifyWait(notifyId, GetKernelExecTimeoutFromEnvConfig()); }
353 :
354 11 : void RtsqA5::NotifyWait(u32 notifyId, u32 timeout)
355 : {
356 11 : BuildA5SqeNotifyWait(streamId_, taskId_, notifyId, timeout, GetCurrSqeBuffer());
357 33 : HCCL_INFO(
358 : "RtsqA5::NotifyWait: streamId %u, taskId %u, notifyId %u, timeout[%u ms]", streamId_, taskId_, notifyId,
359 : timeout);
360 11 : RefreshInfo();
361 11 : }
362 :
363 1 : void RtsqA5::NotifyRecordLoc(u32 notifyId)
364 : {
365 1 : BuildA5SqeNotifyRecord(streamId_, taskId_, notifyId, GetCurrSqeBuffer());
366 3 : HCCL_INFO("RtsqA5::NotifyRecordLoc: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
367 1 : RefreshInfo();
368 1 : }
369 :
370 1 : void RtsqA5::Cnt1toNNotifyWait(u32 notifyId, u32 value)
371 : {
372 1 : BuildA5SqeCnt1toNNotifyWait(streamId_, taskId_, notifyId, value, GetCurrSqeBuffer());
373 3 : HCCL_INFO("RtsqA5::Cnt1toNNotifyWait: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
374 1 : RefreshInfo();
375 1 : }
376 :
377 1 : void RtsqA5::Cnt1toNNotifyRecord(u32 notifyId, u32 value)
378 : {
379 1 : BuildA5SqeCnt1toNNotifyRecord(streamId_, taskId_, notifyId, value, GetCurrSqeBuffer());
380 3 : HCCL_INFO("RtsqA5::Cnt1toNNotifyRecord: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
381 1 : RefreshInfo();
382 1 : }
383 :
384 1 : void RtsqA5::CntNto1NotifyWait(u32 notifyId, u32 value)
385 : {
386 1 : BuildA5SqeCntNto1NotifyWait(streamId_, taskId_, notifyId, value, GetCurrSqeBuffer());
387 3 : HCCL_INFO("RtsqA5::CntNto1NotifyWait: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
388 1 : RefreshInfo();
389 1 : }
390 :
391 1 : void RtsqA5::CntNto1NotifyRecord(u32 notifyId, u32 value)
392 : {
393 1 : BuildA5SqeCntNto1NotifyRecord(streamId_, taskId_, notifyId, value, GetCurrSqeBuffer());
394 3 : HCCL_INFO("RtsqA5::CntNto1NotifyRecord: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
395 1 : RefreshInfo();
396 1 : }
397 :
398 2 : void RtsqA5::SdmaCopy(u64 srcAddr, u64 dstAddr, u32 size, u32 partId)
399 : {
400 : // 不带reduce的拷贝，opcode填0
401 : (void)partId;
402 2 : BuildA5SqeSdmaCopy(streamId_, taskId_, dstAddr, srcAddr, size, RTSQ_A5_PART_ID, 0, GetCurrSqeBuffer());
403 6 : HCCL_INFO(
404 : "RtsqA5::SdmaCopy: streamId %u, taskId %u, srcAddr 0x%llx, dstAddr 0x%llx, size %u", streamId_, taskId_,
405 : srcAddr, dstAddr, size);
406 2 : RefreshInfo();
407 2 : }
408 :
409 : const std::unordered_map<ReduceOp, RtStarsMemcpyAsyncOperationKind, EnumClassHash> ReduceOpToStarsOpKindMap
410 : = {{ReduceOp::SUM, RtStarsMemcpyAsyncOperationKind::RT_STARS_MEMCPY_ASYNC_OP_KIND_ADD},
411 : {ReduceOp::MAX, RtStarsMemcpyAsyncOperationKind::RT_STARS_MEMCPY_ASYNC_OP_KIND_MAX},
412 : {ReduceOp::MIN, RtStarsMemcpyAsyncOperationKind::RT_STARS_MEMCPY_ASYNC_OP_KIND_MIN},
413 : {ReduceOp::EQUAL, RtStarsMemcpyAsyncOperationKind::RT_STARS_MEMCPY_ASYNC_OP_KIND_EQUAL}};
414 :
415 : const std::unordered_map<DataType, RtStarsMemcpyAsyncDataType, EnumClassHash> DataTypeToStarsDataTypeMap
416 : = {{DataType::INT8, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_INT8},
417 : {DataType::INT16, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_INT16},
418 : {DataType::INT32, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_INT32},
419 : {DataType::FP16, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_FP16},
420 : {DataType::FP32, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_FP32},
421 : {DataType::BFP16, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_BFP16}};
422 :
423 3 : void RtsqA5::SdmaReduce(u64 srcAddr, u64 dstAddr, u32 size, u32 partId, const ReduceIn& reduceIn)
424 : {
425 : (void)partId;
426 3 : if (UNLIKELY(
427 : ReduceOpToStarsOpKindMap.find(reduceIn.reduceOp) == ReduceOpToStarsOpKindMap.end()
428 : || DataTypeToStarsDataTypeMap.find(reduceIn.dataType) == DataTypeToStarsDataTypeMap.end())) {
429 3 : THROW<InternalException>(StringFormat(
430 3 : "Sdma does not support reduceOp %s dataType %s", reduceIn.reduceOp.Describe().c_str(),
431 3 : reduceIn.dataType.Describe().c_str()));
432 : }
433 :
434 2 : u8 op = static_cast<u8>(ReduceOpToStarsOpKindMap.at(reduceIn.reduceOp));
435 2 : u8 type = static_cast<u8>(DataTypeToStarsDataTypeMap.at(reduceIn.dataType));
436 :
437 2 : BuildA5SqeSdmaCopy(streamId_, taskId_, dstAddr, srcAddr, size, RTSQ_A5_PART_ID, (op | type), GetCurrSqeBuffer());
438 6 : HCCL_INFO(
439 : "RtsqA5::SdmaReduce: streamId %u, taskId %u, srcAddr 0x%llx, dstAddr 0x%llx, size %u", streamId_, taskId_,
440 : srcAddr, dstAddr, size);
441 2 : RefreshInfo();
442 2 : }
443 :
444 2 : bool RtsqA5::IsRtsqQueueSpaceSufficient()
445 : {
446 : // 判断逻辑与rtsq内部保持一致，rtsq剩余空间需要大于（rtsq挂起的任务数量+本次任务）
447 2 : u32 availableSpace = GetTailToHeadDist();
448 2 : if (availableSpace > pendingSqeCnt + 1) {
449 1 : return true;
450 : }
451 :
452 : // 否则的话，需要再次查询一次head，确认是否是因为head没有更新导致空间不足，如果查询后空间仍然不足，则返回false
453 1 : sqHead_ = QuerySqHead();
454 1 : availableSpace = GetTailToHeadDist();
455 :
456 1 : return (availableSpace > pendingSqeCnt + 1);
457 : }
458 :
459 3 : HcclResult RtsqA5::SetPreStreamSyncReady()
460 : {
461 3 : isPreStreamSync = true;
462 3 : return HCCL_SUCCESS;
463 : }
464 :
465 4 : HcclResult RtsqA5::SetPreStreamSyncFin()
466 : {
467 4 : isPreStreamSync = false;
468 4 : return HCCL_SUCCESS;
469 : }
470 :
471 89 : bool RtsqA5::GetPreStreamSyncStatus() { return isPreStreamSync; }
472 :
473 2 : void RtsqA5::UbDbSend(const UbJettyLiteId& jettyLiteId, u16 piValue)
474 : {
475 : // piValue需要使用u16数据类型，保证自然增长，用于判断是否翻转
476 2 : BuildA5SqeUbDbSend(streamId_, taskId_, jettyLiteId, piValue, GetCurrSqeBuffer());
477 6 : HCCL_INFO(
478 : "RtsqA5::UbDbSend: streamId %u, taskId %u, piValue(UbPi):%u, SqTail(Rtsq Pi):%u", streamId_, taskId_, piValue,
479 : sqTail_);
480 2 : RefreshInfo();
481 2 : }
482 :
483 6 : void RtsqA5::RdmaDbSend(const uint64_t& dbAddr, const uint64_t& dbValue)
484 : {
485 6 : BuildA5SqeRdmaDbSend(streamId_, taskId_, dbAddr, dbValue, GetCurrSqeBuffer());
486 18 : HCCL_INFO(
487 : "RtsqA5::RdmaDbSend: RdmaDbSend streamId %u, taskId %u, Sqe: %s, dbAddr:0x%llx, dbValue:0x%llx, SqTail(Rtsq "
488 : "Pi):%u",
489 : streamId_, taskId_, Bytes2hex(GetCurrSqeBuffer(), RTSQ_SQE_SIZE).c_str(), dbAddr, dbValue, sqTail_);
490 6 : RefreshInfo();
491 6 : }
492 :
493 1 : void RtsqA5::CCoreNotifyWait(u64 waitAddr, u64 curTurnCntAddr, bool last)
494 : {
495 1 : BuildA5SqeCCoreNotifyWait(streamId_, taskId_, waitAddr, curTurnCntAddr, last, GetCurrSqeBuffer());
496 3 : HCCL_INFO(
497 : "RtsqA5::CCoreNotifyWait: streamId %u, taskId %u, waitAddr %llu, curTurnCntAddr %llu, last %d", streamId_,
498 : taskId_, waitAddr, curTurnCntAddr, last);
499 1 : RefreshInfo();
500 1 : }
501 :
502 1 : void RtsqA5::CCoreNotifyRecord(u64 recordAddr, u64 curTurnCntAddr)
503 : {
504 1 : BuildA5SqeCCoreNotifyRecord(streamId_, taskId_, recordAddr, curTurnCntAddr, GetCurrSqeBuffer());
505 3 : HCCL_INFO(
506 : "RtsqA5::CCoreNotifyRecord: streamId %u, taskId %u, recordAddr %llu, curTurnCntAddr %llu", streamId_, taskId_,
507 : recordAddr, curTurnCntAddr);
508 1 : RefreshInfo();
509 1 : }
510 :
511 0 : void RtsqA5::P2PWriteValue(u64 remoteAddr, u32 writeValue)
512 : {
513 0 : BuildA5SqeP2pWriteValue(streamId_, taskId_, remoteAddr, writeValue, GetCurrSqeBuffer());
514 0 : HCCL_INFO(
515 : "RtsqA5::P2PWriteValue: streamId %u, taskId %u, remoteAddr %llu, writeValue %u", streamId_, taskId_, remoteAddr,
516 : writeValue);
517 0 : RefreshInfo();
518 0 : }
519 :
520 0 : HcclResult RtsqA5::GetLastStreamIdAndTaskId(uint16_t& streamId, uint16_t& taskId) const
521 : {
522 0 : if (pendingSqeCnt > 0) {
523 0 : const u8* lastSqe = locBuf + (pendingSqeCnt - 1U) * RTSQ_SQE_SIZE;
524 0 : auto* sqe = reinterpret_cast<const Rt91095StarsNotifySqe*>(lastSqe);
525 0 : streamId = sqe->header.rtStreamId;
526 0 : taskId = sqe->header.taskId;
527 0 : HCCL_INFO(
528 : "[%s] from pending, pendingSqeCnt[%u], sqId[%u], streamId[%u], taskId[%u].", __func__, pendingSqeCnt, sqId_,
529 : streamId, taskId);
530 0 : return HCCL_SUCCESS;
531 : }
532 0 : const u32 lastIdx = (sqTail_ + sqDepth_ - 1U) % sqDepth_;
533 0 : HCCL_INFO(
534 : "[%s] from rtsq, sqId[%u], sqTail[%u], sqDepth[%u], lastIdx[%u].", __func__, sqId_, sqTail_, sqDepth_, lastIdx);
535 0 : return GetStreamIdAndTaskIdBySqIdx(lastIdx, streamId, taskId);
536 : }
537 : } // namespace Hccl
|
